From 71e81b6e0c230ed16f40c6c1b962139d2558bb22 Mon Sep 17 00:00:00 2001 From: xiaohanzi Date: Tue, 9 Jun 2020 23:40:43 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0=E9=98=BF=E9=87=8C=E4=BA=91ro?= =?UTF-8?q?cketmq?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/main/resources/application-dev.yml | 9 +-- .../src/main/resources/application-dev.yml | 8 +- .../src/main/resources/application-dev.yml | 8 +- .../src/main/resources/application-dev.yml | 8 +- .../mq/impl/AliyunRocketMqConsumer.java | 73 +++++++----------- .../src/main/resources/application-dev.yml | 4 +- .../src/main/resources/application-dev.yml | 8 +- .../src/main/resources/application-dev.yml | 8 +- .../impl/AliyunRocketMqMessageProducer.java | 74 +++++++++++-------- 9 files changed, 96 insertions(+), 104 deletions(-) diff --git a/mallinkAdmin/src/main/resources/application-dev.yml b/mallinkAdmin/src/main/resources/application-dev.yml index 0f74778dd..588874364 100644 --- a/mallinkAdmin/src/main/resources/application-dev.yml +++ b/mallinkAdmin/src/main/resources/application-dev.yml @@ -73,12 +73,11 @@ spring: publisher-confirms: true publisher-returns: false virtual-host: / - aliyunRocketmq: - accessKeyId: LTAI4G2hFwCBVhxCFbV7nBxf - accessKeySecret: LTAI4G2hFwCBVhxCFbV7nBxf - groupId: GID_P_1 - namesrvAddr: http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 + accessKeyId: "LTAI4G7ixY4AhvM35F8o3W3V" + accessKeySecret: "VfWqGb83qIQrS9us45utskl8itd7ry" + groupId: "GID_P_1" + namesrvAddr: "http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080" flyway: enabled: false diff --git a/mallinkBApi/src/main/resources/application-dev.yml b/mallinkBApi/src/main/resources/application-dev.yml index 5a8a7f6f1..3e922e7b6 100644 --- a/mallinkBApi/src/main/resources/application-dev.yml +++ b/mallinkBApi/src/main/resources/application-dev.yml @@ -73,10 +73,10 @@ spring: publisher-returns: false virtual-host: / aliyunRocketmq: - accessKeyId: LTAI4G2hFwCBVhxCFbV7nBxf - accessKeySecret: LTAI4G2hFwCBVhxCFbV7nBxf - groupId: GID_P_1 - namesrvAddr: http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 + accessKeyId: "LTAI4G7ixY4AhvM35F8o3W3V" + accessKeySecret: "VfWqGb83qIQrS9us45utskl8itd7ry" + groupId: "GID_P_1" + namesrvAddr: "http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080" aws: clientRegion: cn-northwest-1 diff --git a/mallinkCApi/src/main/resources/application-dev.yml b/mallinkCApi/src/main/resources/application-dev.yml index 5057809ef..58dd6618e 100644 --- a/mallinkCApi/src/main/resources/application-dev.yml +++ b/mallinkCApi/src/main/resources/application-dev.yml @@ -73,10 +73,10 @@ spring: publisher-returns: false virtual-host: / aliyunRocketmq: - accessKeyId: LTAI4G2hFwCBVhxCFbV7nBxf - accessKeySecret: LTAI4G2hFwCBVhxCFbV7nBxf - groupId: GID_P_1 - namesrvAddr: http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 + accessKeyId: "LTAI4G7ixY4AhvM35F8o3W3V" + accessKeySecret: "VfWqGb83qIQrS9us45utskl8itd7ry" + groupId: "GID_P_1" + namesrvAddr: "http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080" aws: clientRegion: cn-northwest-1 diff --git a/mallinkCallback/src/main/resources/application-dev.yml b/mallinkCallback/src/main/resources/application-dev.yml index 7c49dc492..6a534086e 100644 --- a/mallinkCallback/src/main/resources/application-dev.yml +++ b/mallinkCallback/src/main/resources/application-dev.yml @@ -75,10 +75,10 @@ spring: virtual-host: / aliyunRocketmq: - accessKeyId: LTAI4G2hFwCBVhxCFbV7nBxf - accessKeySecret: LTAI4G2hFwCBVhxCFbV7nBxf - groupId: GID_P_1 - namesrvAddr: http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 + accessKeyId: "LTAI4G7ixY4AhvM35F8o3W3V" + accessKeySecret: "VfWqGb83qIQrS9us45utskl8itd7ry" + groupId: "GID_P_1" + namesrvAddr: "http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080" aws: clientRegion: cn-northwest-1 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 9a6cf8687..cf684f2d4 100644 --- a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java +++ b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java @@ -1,16 +1,12 @@ package com.iformall.mq.impl; -import java.util.HashMap; -import java.util.Map; import java.util.Properties; import javax.annotation.PostConstruct; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Profile; +import org.springframework.stereotype.Service; import com.aliyun.openservices.ons.api.Action; import com.aliyun.openservices.ons.api.ConsumeContext; @@ -20,7 +16,6 @@ import com.aliyun.openservices.ons.api.MessageListener; import com.aliyun.openservices.ons.api.ONSFactory; import com.aliyun.openservices.ons.api.PropertyKeyConst; import com.aliyun.openservices.ons.api.bean.ConsumerBean; -import com.aliyun.openservices.ons.api.bean.Subscription; import com.iformall.enums.EnumMsgMqTopic; import com.iformall.mq.MQConfig; import com.iformall.mq.MqBaseConsumer; @@ -28,7 +23,7 @@ import com.iformall.mq.MqBaseConsumer; import lombok.extern.slf4j.Slf4j; @Slf4j -@Configuration +@Service @Profile(MQConfig.Impl.ALIYUN_ROCKET_MQ) public class AliyunRocketMqConsumer extends MqBaseConsumer { @@ -44,46 +39,34 @@ public class AliyunRocketMqConsumer extends MqBaseConsumer { @Value("${spring.aliyunRocketmq.namesrvAddr}") private String namesrvAddr; - @Bean(initMethod = "start", destroyMethod = "shutdown") - public ConsumerBean buildConsumer() { - ConsumerBean consumerBean = new ConsumerBean(); - //配置文件 - Properties properties = new Properties(); - properties.setProperty(PropertyKeyConst.AccessKey, accessKeyId); - properties.setProperty(PropertyKeyConst.SecretKey, accessKeySecret); - properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, namesrvAddr); + private static Consumer consumer = null; + + @PostConstruct + public void init() { + Properties properties = new Properties(); + // 您在控制台创建的 Group ID + properties.put(PropertyKeyConst.GROUP_ID, groupId); + // AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建 + properties.put(PropertyKeyConst.AccessKey, accessKeyId); + // SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建 + properties.put(PropertyKeyConst.SecretKey, accessKeySecret); + // 设置 TCP 接入域名,到控制台的实例基本信息中查看 + properties.put(PropertyKeyConst.NAMESRV_ADDR,namesrvAddr); + // 集群订阅方式(默认) + // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING); + // 广播订阅方式 + // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING); + consumer = ONSFactory.createConsumer(properties); - properties.setProperty(PropertyKeyConst.GROUP_ID, groupId); - //将消费者线程数固定为20个 20为默认值 - properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "20"); - consumerBean.setProperties(properties); - //订阅关系 - Map subscriptionTable = new HashMap(); - Subscription subscription = new Subscription(); - subscription.setTopic(EnumMsgMqTopic.DEFAULT.getCode()); - subscription.setExpression("*"); - subscriptionTable.put(subscription, new MessageListener() { - @Override - public Action consume(Message message, ConsumeContext context) { - log.info("Receive rocket msg: " + message); - try { - doMessage(new String(message.getBody())); - return Action.CommitMessage; - } catch (Exception e) { - //消费失败 - return Action.ReconsumeLater; - } - } - + consumer.subscribe(EnumMsgMqTopic.DEFAULT.getCode(), "*", new MessageListener() { //订阅多个 Tag + public Action consume(Message message, ConsumeContext context) { + System.out.println("Receive: " + message); + doMessage(new String(message.getBody())); + return Action.CommitMessage; + } }); - //订阅多个topic如上面设置 - - consumerBean.setSubscriptionTable(subscriptionTable); - return consumerBean; - } - - - + } + // @PostConstruct // public void init() { // Properties properties = new Properties(); diff --git a/mallinkMQConsumer/src/main/resources/application-dev.yml b/mallinkMQConsumer/src/main/resources/application-dev.yml index 7c2e5b9ee..1fbada993 100644 --- a/mallinkMQConsumer/src/main/resources/application-dev.yml +++ b/mallinkMQConsumer/src/main/resources/application-dev.yml @@ -73,8 +73,8 @@ spring: publisher-returns: false virtual-host: / aliyunRocketmq: - accessKeyId: "LTAI4G2hFwCBVhxCFbV7nBxf" - accessKeySecret: "LTAI4G2hFwCBVhxCFbV7nBxf" + accessKeyId: "LTAI4G7ixY4AhvM35F8o3W3V" + accessKeySecret: "VfWqGb83qIQrS9us45utskl8itd7ry" groupId: "GID_P_1" namesrvAddr: "http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080" diff --git a/mallinkPosApi/src/main/resources/application-dev.yml b/mallinkPosApi/src/main/resources/application-dev.yml index 7bbbd9673..54c956f1b 100644 --- a/mallinkPosApi/src/main/resources/application-dev.yml +++ b/mallinkPosApi/src/main/resources/application-dev.yml @@ -73,10 +73,10 @@ spring: publisher-returns: false virtual-host: / aliyunRocketmq: - accessKeyId: LTAI4G2hFwCBVhxCFbV7nBxf - accessKeySecret: LTAI4G2hFwCBVhxCFbV7nBxf - groupId: GID_P_1 - namesrvAddr: http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 + accessKeyId: "LTAI4G7ixY4AhvM35F8o3W3V" + accessKeySecret: "VfWqGb83qIQrS9us45utskl8itd7ry" + groupId: "GID_P_1" + namesrvAddr: "http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080" aws: clientRegion: cn-northwest-1 diff --git a/mallinkSchedule/src/main/resources/application-dev.yml b/mallinkSchedule/src/main/resources/application-dev.yml index 8fb701e2b..6a124f9cb 100644 --- a/mallinkSchedule/src/main/resources/application-dev.yml +++ b/mallinkSchedule/src/main/resources/application-dev.yml @@ -73,10 +73,10 @@ spring: publisher-returns: false virtual-host: / aliyunRocketmq: - accessKeyId: LTAI4G2hFwCBVhxCFbV7nBxf - accessKeySecret: LTAI4G2hFwCBVhxCFbV7nBxf - groupId: GID_P_1 - namesrvAddr: http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 + accessKeyId: "LTAI4G7ixY4AhvM35F8o3W3V" + accessKeySecret: "VfWqGb83qIQrS9us45utskl8itd7ry" + groupId: "GID_P_1" + namesrvAddr: "http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080" aws: clientRegion: cn-northwest-1 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 85d9ccca5..bde7e327f 100644 --- a/mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java +++ b/mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java @@ -1,5 +1,6 @@ package com.iformall.mq.impl; +import java.util.Date; import java.util.Properties; import javax.annotation.PostConstruct; @@ -11,10 +12,10 @@ import org.springframework.context.annotation.Profile; import org.springframework.stereotype.Service; import com.aliyun.openservices.ons.api.Message; +import com.aliyun.openservices.ons.api.ONSFactory; +import com.aliyun.openservices.ons.api.Producer; import com.aliyun.openservices.ons.api.PropertyKeyConst; import com.aliyun.openservices.ons.api.SendResult; -import com.aliyun.openservices.ons.api.bean.OrderProducerBean; -import com.aliyun.openservices.ons.api.exception.ONSClientException; import com.iformall.domain.po.msg.BaseMsg; import com.iformall.mq.MQConfig; import com.iformall.mq.MqBaseProducer; @@ -44,55 +45,64 @@ public class AliyunRocketMqMessageProducer implements MqBaseProducer { @Value("${spring.aliyunRocketmq.namesrvAddr}") private String namesrvAddr; - //顺序消息的Producer 已经注册到了spring容器中,后面需要使用时可以直接注入到其它类中 - private static OrderProducerBean orderProducer; - - //@Autowired //private WxMsgRecordMapper wxMsgRecordMapper; //@Autowired //private WxMsgRecordService wxMsgRecordService; + private static Producer producer = null; + @PostConstruct public void init() { - orderProducer = new OrderProducerBean(); Properties properties = new Properties(); - properties.setProperty(PropertyKeyConst.AccessKey, accessKeyId); - properties.setProperty(PropertyKeyConst.SecretKey, accessKeySecret); - properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, namesrvAddr); - orderProducer.setProperties(properties); + properties.setProperty(PropertyKeyConst.GROUP_ID, groupId); + // AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建 + properties.put(PropertyKeyConst.AccessKey,accessKeyId); + // SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建 + properties.put(PropertyKeyConst.SecretKey, accessKeySecret); + //设置发送超时时间,单位毫秒 + properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000"); + // 设置 TCP 接入域名,到控制台的实例基本信息中查看 + properties.put(PropertyKeyConst.NAMESRV_ADDR,namesrvAddr); + producer = ONSFactory.createProducer(properties); + // 在发送消息前,必须调用 start 方法来启动 Producer,只需调用一次即可 + producer.start(); } @Override public void sendMessage(BaseMsg data, String topic, String tags, String keys) { - String shardingKey = "OrderedKey"; //循环发送消息 - Message msg = new Message( // - // Message所属的Topic - topic, - // Message Tag 可理解为Gmail中的标签,对消息进行再归类,方便Consumer指定过滤条件在MQ服务器过滤 - tags, - // Message Body 可以是任何二进制形式的数据, MQ不做任何干预 - // 需要Producer与Consumer协商好一致的序列化和反序列化方式 - JsonUtil.obj2Json(data).getBytes()); - // 设置代表消息的业务关键属性,请尽可能全局唯一 - // 以方便您在无法正常收到消息情况下,可通过MQ 控制台查询消息并补发 + Message msg = new Message( // + // Message 所属的 Topic + topic, + // Message Tag 可理解为 Gmail 中的标签,对消息进行再归类,方便 Consumer 指定过滤条件在 MQ 服务器过滤 + tags, + // Message Body 可以是任何二进制形式的数据, MQ 不做任何干预, + // 需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式 + JsonUtil.obj2Json(data).getBytes()); + // 设置代表消息的业务关键属性,请尽可能全局唯一。 + // 以方便您在无法正常收到消息情况下,可通过阿里云服务器管理控制台查询消息并补发 // 注意:不设置也不会影响消息正常收发 msg.setKey(keys); - // 发送消息,只要不抛异常就是成功 + try { - //data.setUuid(UUID.randomUUID().toString()); - //wxMsgRecordService.save(data); - SendResult sendResult = orderProducer.send(msg, shardingKey); - assert sendResult != null; - System.out.println(sendResult); - } catch (ONSClientException e) { - System.out.println("发送失败"); - //出现异常意味着发送失败,为了避免消息丢失,建议缓存该消息然后进行重试。 + SendResult sendResult = producer.send(msg); + // 同步发送消息,只要不抛异常就是成功 + if (sendResult != null) { + System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId()); + } } - + catch (Exception e) { + // 消息发送失败,需要进行重试处理,可重新发送这条消息或持久化这条数据进行补偿处理 + System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic()); + e.printStackTrace(); + } + + // 在应用退出前,销毁 Producer 对象 + // 注意:如果不销毁也没有问题 + //producer.shutdown(); // try { // data.setUuid(UUID.randomUUID().toString());