From 2f343f1495b48e8a68025e453d5d4308d62d4381 Mon Sep 17 00:00:00 2001 From: xiaohanzi Date: Wed, 10 Jun 2020 09:29:55 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A7=A3=E5=86=B3=E9=98=BF=E9=87=8C=E4=BA=91ro?= =?UTF-8?q?cketmq=E6=8E=A5=E5=85=A5bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../mq/impl/AliyunRocketMqConsumer.java | 32 ++----------------- .../impl/AliyunRocketMqMessageProducer.java | 32 ++----------------- 2 files changed, 5 insertions(+), 59 deletions(-) diff --git a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java index cf684f2d4..f14a28f11 100644 --- a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java +++ b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java @@ -65,35 +65,9 @@ public class AliyunRocketMqConsumer extends MqBaseConsumer { return Action.CommitMessage; } }); + + consumer.start(); + log.info(" aliyunrocketMq consumer start success!"); } -// @PostConstruct -// public void init() { -// Properties properties = new Properties(); -// // 您在控制台创建的 Group ID -// properties.put(PropertyKeyConst.GROUP_ID, consumerGroupId); -// // 鉴权用 AccessKeyId,在阿里云服务器管理控制台创建 -// properties.put(PropertyKeyConst.AccessKey, accessKeyId); -// // 鉴权用 AccessKeySecret,在阿里云服务器管理控制台创建 -// properties.put(PropertyKeyConst.SecretKey, accessKeySecret); -// // 设置 TCP 接入域名,进入控制台的实例详情页面,在页面上方选择实例后,在实例信息中的“获取接入点信息”区域查看 -// properties.put(PropertyKeyConst.NAMESRV_ADDR,namesrvAddr); -// -// Consumer consumer = ONSFactory.createConsumer(properties); -// consumer.subscribe("topic-1", "*", new MessageListener() { -// public Action consume(Message message, ConsumeContext context) { -// System.out.println("Receive: " + message); -// String msg = new String(message.getBody()); -// doMessage(msg); -// return Action.CommitMessage; -// } -// }); -// consumer.start(); -// System.out.println("Consumer Started"); -// } - - -// public void onMessage(String message) { -// doMessage(message); -// } } diff --git a/mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java b/mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java index bde7e327f..6557542f6 100644 --- a/mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java +++ b/mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java @@ -27,7 +27,7 @@ import com.iformall.utils.JsonUtil; * @Date: 2019-01-31 * @Description: RocketMQ消息生产者 */ -@Service +//@Service @Profile(MQConfig.Impl.ALIYUN_ROCKET_MQ) public class AliyunRocketMqMessageProducer implements MqBaseProducer { private final Logger log = LoggerFactory.getLogger(this.getClass()); @@ -53,7 +53,7 @@ public class AliyunRocketMqMessageProducer implements MqBaseProducer { private static Producer producer = null; - @PostConstruct + // @PostConstruct public void init() { Properties properties = new Properties(); properties.setProperty(PropertyKeyConst.GROUP_ID, groupId); @@ -103,33 +103,5 @@ public class AliyunRocketMqMessageProducer implements MqBaseProducer { // 在应用退出前,销毁 Producer 对象 // 注意:如果不销毁也没有问题 //producer.shutdown(); - -// try { -// data.setUuid(UUID.randomUUID().toString()); -// wxMsgRecordService.save(data); -// Message msg = new Message( // -// // 在控制台创建的 Topic,即该消息所属的 Topic 名称 -// topic, -// // Message Tag, -// // 可理解为 Gmail 中的标签,对消息进行再归类,方便 Consumer 指定过滤条件在消息队列 RocketMQ 版服务器过滤 -// tags, -// // Message Body -// // 任何二进制形式的数据,消息队列 RocketMQ 版不做任何干预, -// // 需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式 -// JsonUtil.obj2Json(data).getBytes()); -// // 设置代表消息的业务关键属性,请尽可能全局唯一,以方便您在无法正常收到消息情况下,可通过控制台查询消息并补发 -// // 注意:不设置也不会影响消息正常收发 -// msg.setKey(keys); -// // 发送消息,只要不抛异常就是成功 -// // 打印 Message ID,以便用于消息发送状态查询 -// SendResult sendResult = getProducer(producerGroupId).send(msg); -// //System.out.println("Send Message success. Message ID is: " + sendResult.getMessageId()); -// data.setMsgStatus(EnumMsgRecordStatus.SEND_SUCC.getCode()); -// wxMsgRecordMapper.updateByBaseMsg(data); -// } catch (Exception e) { -// log.error("Message Producer: Send Message Error ", e); -// } - } - }