From 1fe96beb84a89af699eacebea10d6f744c8c0d91 Mon Sep 17 00:00:00 2001 From: xiaohanzi Date: Wed, 3 Jun 2020 23:49:17 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8E=A5=E5=85=A5=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 | 8 +- .../src/main/resources/application.yml | 32 ++--- .../src/main/resources/application-dev.yml | 9 +- .../src/main/resources/application.yml | 18 +-- .../src/main/resources/application-dev.yml | 9 +- .../src/main/resources/application.yml | 18 +-- .../src/main/resources/application-dev.yml | 9 +- .../src/main/resources/application.yml | 32 ++--- .../mq/impl/AliyunRocketMqConsumer.java | 71 +++++++++++ .../src/main/resources/application-dev.yml | 9 +- .../src/main/resources/application.yml | 32 ++--- .../src/main/resources/application-dev.yml | 9 +- .../src/main/resources/application-dev.yml | 9 +- .../src/main/resources/application.yml | 32 ++--- mallinkService/pom.xml | 7 ++ .../main/java/com/iformall/mq/MQConfig.java | 1 + .../impl/AliyunRocketMqMessageProducer.java | 112 ++++++++++++++++++ .../src/main/resources/application-dev.yml | 9 +- .../src/main/resources/application.yml | 32 ++--- 19 files changed, 345 insertions(+), 113 deletions(-) create mode 100644 mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java create mode 100644 mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java diff --git a/mallinkAdmin/src/main/resources/application-dev.yml b/mallinkAdmin/src/main/resources/application-dev.yml index b30ac38e0..b5a499ede 100644 --- a/mallinkAdmin/src/main/resources/application-dev.yml +++ b/mallinkAdmin/src/main/resources/application-dev.yml @@ -1,6 +1,7 @@ spring: profiles: - include: rabbitMQ + include: aliyunRocketMQ + #include: rabbitMQ # JDBC datasource: url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true&allowMultiQueries=true @@ -72,6 +73,11 @@ spring: publisher-confirms: true publisher-returns: false virtual-host: / + aliyunRocketmq: + accessKeyId:LTAI4G2hFwCBVhxCFbV7nBxf + accessKeySecret:LTAI4G2hFwCBVhxCFbV7nBxf + producerGroupId:GID_P_1 + namesrvAddr:http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 flyway: enabled: false diff --git a/mallinkAdmin/src/main/resources/application.yml b/mallinkAdmin/src/main/resources/application.yml index 836b08aee..0ea90c067 100644 --- a/mallinkAdmin/src/main/resources/application.yml +++ b/mallinkAdmin/src/main/resources/application.yml @@ -21,22 +21,22 @@ spring: cache-names: redis_cache #缓存的名字(可以不指定) redis: time-to-live: 60000ms #很重要,缓存的有效时间,以便缓存的过期(单位为毫秒) - rocketmq: - nameServer: 202.165.179.86:9876 - producer: - retry-times-when-send-async-failed: 0 - send-msg-timeout: 300000 - compress-msg-body-over-howmuch: 4096 - max-message-size: 4194304 - retry-another-broker-when-not-store-ok: false - retry-times-when-send-failed: 2 - rabbitmq: - host: 202.165.179.86 - port: 5672 - username: guest - password: guest - publisher-confirms: true - virtual-host: / +# rocketmq: +# nameServer: 202.165.179.86:9876 +# producer: +# retry-times-when-send-async-failed: 0 +# send-msg-timeout: 300000 +# compress-msg-body-over-howmuch: 4096 +# max-message-size: 4194304 +# retry-another-broker-when-not-store-ok: false +# retry-times-when-send-failed: 2 +# rabbitmq: +# host: 202.165.179.86 +# port: 5672 +# username: guest +# password: guest +# publisher-confirms: true +# virtual-host: / # FLOWABLE flowable: async-executor-activate: false diff --git a/mallinkBApi/src/main/resources/application-dev.yml b/mallinkBApi/src/main/resources/application-dev.yml index e710131bf..b3bcf046d 100644 --- a/mallinkBApi/src/main/resources/application-dev.yml +++ b/mallinkBApi/src/main/resources/application-dev.yml @@ -1,6 +1,7 @@ spring: profiles: - include: rabbitMQ + include: aliyunRocketMQ + #include: rabbitMQ # JDBC datasource: url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true @@ -71,7 +72,11 @@ spring: publisher-confirms: true publisher-returns: false virtual-host: / - + aliyunRocketmq: + accessKeyId:LTAI4G2hFwCBVhxCFbV7nBxf + accessKeySecret:LTAI4G2hFwCBVhxCFbV7nBxf + producerGroupId:GID_P_1 + namesrvAddr:http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 aws: clientRegion: cn-northwest-1 bucketName: iformall-net diff --git a/mallinkBApi/src/main/resources/application.yml b/mallinkBApi/src/main/resources/application.yml index 0b1867eca..8f0d421b9 100644 --- a/mallinkBApi/src/main/resources/application.yml +++ b/mallinkBApi/src/main/resources/application.yml @@ -12,15 +12,15 @@ spring: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 default-property-inclusion: non_null - rocketmq: - nameServer: 127.0.0.1:9876 - producer: - retry-times-when-send-async-failed: 0 - send-msg-timeout: 300000 - compress-msg-body-over-howmuch: 4096 - max-message-size: 4194304 - retry-another-broker-when-not-store-ok: false - retry-times-when-send-failed: 2 +# rocketmq: +# nameServer: 127.0.0.1:9876 +# producer: +# retry-times-when-send-async-failed: 0 +# send-msg-timeout: 300000 +# compress-msg-body-over-howmuch: 4096 +# max-message-size: 4194304 +# retry-another-broker-when-not-store-ok: false +# retry-times-when-send-failed: 2 # MybatisPlus mybatis-plus: diff --git a/mallinkCApi/src/main/resources/application-dev.yml b/mallinkCApi/src/main/resources/application-dev.yml index a9a0a049d..e454699a0 100644 --- a/mallinkCApi/src/main/resources/application-dev.yml +++ b/mallinkCApi/src/main/resources/application-dev.yml @@ -1,6 +1,7 @@ spring: profiles: - include: rabbitMQ + include: aliyunRocketMQ + #include: rabbitMQ # JDBC datasource: url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true @@ -71,7 +72,11 @@ spring: publisher-confirms: true publisher-returns: false virtual-host: / - + aliyunRocketmq: + accessKeyId:LTAI4G2hFwCBVhxCFbV7nBxf + accessKeySecret:LTAI4G2hFwCBVhxCFbV7nBxf + producerGroupId:GID_P_1 + namesrvAddr:http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 aws: clientRegion: cn-northwest-1 bucketName: iformall-net diff --git a/mallinkCApi/src/main/resources/application.yml b/mallinkCApi/src/main/resources/application.yml index 075505606..324f91cde 100644 --- a/mallinkCApi/src/main/resources/application.yml +++ b/mallinkCApi/src/main/resources/application.yml @@ -12,15 +12,15 @@ spring: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 default-property-inclusion: non_null - rocketmq: - nameServer: 127.0.0.1:9876 - producer: - retry-times-when-send-async-failed: 0 - send-msg-timeout: 300000 - compress-msg-body-over-howmuch: 4096 - max-message-size: 4194304 - retry-another-broker-when-not-store-ok: false - retry-times-when-send-failed: 2 +# rocketmq: +# nameServer: 127.0.0.1:9876 +# producer: +# retry-times-when-send-async-failed: 0 +# send-msg-timeout: 300000 +# compress-msg-body-over-howmuch: 4096 +# max-message-size: 4194304 +# retry-another-broker-when-not-store-ok: false +# retry-times-when-send-failed: 2 # MybatisPlus mybatis-plus: diff --git a/mallinkCallback/src/main/resources/application-dev.yml b/mallinkCallback/src/main/resources/application-dev.yml index 81a8c909f..16c34f754 100644 --- a/mallinkCallback/src/main/resources/application-dev.yml +++ b/mallinkCallback/src/main/resources/application-dev.yml @@ -1,6 +1,7 @@ spring: profiles: - include: rabbitMQ + include: aliyunRocketMQ + #include: rabbitMQ # JDBC datasource: url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true @@ -72,7 +73,11 @@ spring: publisher-confirms: true publisher-returns: false virtual-host: / - + aliyunRocketmq: + accessKeyId:LTAI4G2hFwCBVhxCFbV7nBxf + accessKeySecret:LTAI4G2hFwCBVhxCFbV7nBxf + producerGroupId:GID_P_1 + namesrvAddr:http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 aws: clientRegion: cn-northwest-1 bucketName: iformall-net diff --git a/mallinkCallback/src/main/resources/application.yml b/mallinkCallback/src/main/resources/application.yml index 0d1863f82..08133c305 100644 --- a/mallinkCallback/src/main/resources/application.yml +++ b/mallinkCallback/src/main/resources/application.yml @@ -21,22 +21,22 @@ spring: cache-names: redis_cache #缓存的名字(可以不指定) redis: time-to-live: 60000ms #很重要,缓存的有效时间,以便缓存的过期(单位为毫秒) - rocketmq: - nameServer: 202.165.179.86:9876 - producer: - retry-times-when-send-async-failed: 0 - send-msg-timeout: 300000 - compress-msg-body-over-howmuch: 4096 - max-message-size: 4194304 - retry-another-broker-when-not-store-ok: false - retry-times-when-send-failed: 2 - rabbitmq: - host: 202.165.179.86 - port: 5672 - username: guest - password: guest - publisher-confirms: true - virtual-host: / +# rocketmq: +# nameServer: 202.165.179.86:9876 +# producer: +# retry-times-when-send-async-failed: 0 +# send-msg-timeout: 300000 +# compress-msg-body-over-howmuch: 4096 +# max-message-size: 4194304 +# retry-another-broker-when-not-store-ok: false +# retry-times-when-send-failed: 2 +# rabbitmq: +# host: 202.165.179.86 +# port: 5672 +# username: guest +# password: guest +# publisher-confirms: true +# virtual-host: / aop: proxy-target-class: true auto: true diff --git a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java new file mode 100644 index 000000000..c4cec5c08 --- /dev/null +++ b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java @@ -0,0 +1,71 @@ +package com.iformall.mq.impl; + +import java.util.Properties; + +import javax.annotation.PostConstruct; + +import org.springframework.beans.factory.annotation.Value; +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; +import com.aliyun.openservices.ons.api.Consumer; +import com.aliyun.openservices.ons.api.Message; +import com.aliyun.openservices.ons.api.MessageListener; +import com.aliyun.openservices.ons.api.ONSFactory; +import com.aliyun.openservices.ons.api.PropertyKeyConst; +import com.iformall.mq.MQConfig; +import com.iformall.mq.MqBaseConsumer; + +import lombok.extern.slf4j.Slf4j; + +@Slf4j +@Service +@Profile(MQConfig.Impl.ALIYUN_ROCKET_MQ) +public class AliyunRocketMqConsumer extends MqBaseConsumer { + + + @Value("${spring.aliyunRocketmq.accessKeyId}") + private String accessKeyId; + + @Value("${spring.aliyunRocketmq.accessKeySecret}") + private String accessKeySecret; + + @Value("${spring.aliyunRocketmq.consumerGroupId}") + private String consumerGroupId; + + @Value("${spring.aliyunRocketmq.namesrvAddr}") + private String namesrvAddr; + + + @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/mallinkMQConsumer/src/main/resources/application-dev.yml b/mallinkMQConsumer/src/main/resources/application-dev.yml index b1eb509a7..5608d2746 100644 --- a/mallinkMQConsumer/src/main/resources/application-dev.yml +++ b/mallinkMQConsumer/src/main/resources/application-dev.yml @@ -1,6 +1,7 @@ spring: profiles: - include: rabbitMQ + include: aliyunRocketMQ + #include: rabbitMQ # JDBC datasource: url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true @@ -71,7 +72,11 @@ spring: publisher-confirms: true publisher-returns: false virtual-host: / - + aliyunRocketmq: + accessKeyId:LTAI4G2hFwCBVhxCFbV7nBxf + accessKeySecret:IIRfygUsnNNP1KaF3OTRdo0z9KzkbF + consumerGroupId:GID_C_1 + namesrvAddr://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 aws: clientRegion: cn-northwest-1 bucketName: iformall-net diff --git a/mallinkMQConsumer/src/main/resources/application.yml b/mallinkMQConsumer/src/main/resources/application.yml index 2d88312a8..8811981f5 100644 --- a/mallinkMQConsumer/src/main/resources/application.yml +++ b/mallinkMQConsumer/src/main/resources/application.yml @@ -21,22 +21,22 @@ spring: cache-names: redis_cache #缓存的名字(可以不指定) redis: time-to-live: 60000ms #很重要,缓存的有效时间,以便缓存的过期(单位为毫秒) - rocketmq: - nameServer: 127.0.0.1:9876 - producer: - retry-times-when-send-async-failed: 0 - send-msg-timeout: 300000 - compress-msg-body-over-howmuch: 4096 - max-message-size: 4194304 - retry-another-broker-when-not-store-ok: false - retry-times-when-send-failed: 2 - rabbitmq: - host: 202.165.179.86 - port: 5672 - username: guest - password: guest - publisher-confirms: true - virtual-host: / +# rocketmq: +# nameServer: 127.0.0.1:9876 +# producer: +# retry-times-when-send-async-failed: 0 +# send-msg-timeout: 300000 +# compress-msg-body-over-howmuch: 4096 +# max-message-size: 4194304 +# retry-another-broker-when-not-store-ok: false +# retry-times-when-send-failed: 2 +# rabbitmq: +# host: 202.165.179.86 +# port: 5672 +# username: guest +# password: guest +# publisher-confirms: true +# virtual-host: / # MybatisPlus diff --git a/mallinkPosApi/src/main/resources/application-dev.yml b/mallinkPosApi/src/main/resources/application-dev.yml index dcb892a70..54b0e1ca5 100644 --- a/mallinkPosApi/src/main/resources/application-dev.yml +++ b/mallinkPosApi/src/main/resources/application-dev.yml @@ -1,6 +1,7 @@ spring: profiles: - include: rabbitMQ + include: aliyunRocketMQ + #include: rabbitMQ # JDBC datasource: url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true @@ -71,7 +72,11 @@ spring: publisher-confirms: true publisher-returns: false virtual-host: / - + aliyunRocketmq: + accessKeyId:LTAI4G2hFwCBVhxCFbV7nBxf + accessKeySecret:LTAI4G2hFwCBVhxCFbV7nBxf + producerGroupId:GID_P_1 + namesrvAddr:http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 aws: clientRegion: cn-northwest-1 bucketName: iformall-net diff --git a/mallinkSchedule/src/main/resources/application-dev.yml b/mallinkSchedule/src/main/resources/application-dev.yml index dd991b315..ce1c728a3 100644 --- a/mallinkSchedule/src/main/resources/application-dev.yml +++ b/mallinkSchedule/src/main/resources/application-dev.yml @@ -1,6 +1,7 @@ spring: profiles: - include: rabbitMQ + include: aliyunRocketMQ + #include: rabbitMQ # JDBC datasource: url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true @@ -71,7 +72,11 @@ spring: publisher-confirms: true publisher-returns: false virtual-host: / - + aliyunRocketmq: + accessKeyId:LTAI4G2hFwCBVhxCFbV7nBxf + accessKeySecret:LTAI4G2hFwCBVhxCFbV7nBxf + producerGroupId:GID_P_1 + namesrvAddr:http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 aws: clientRegion: cn-northwest-1 bucketName: iformall-net diff --git a/mallinkSchedule/src/main/resources/application.yml b/mallinkSchedule/src/main/resources/application.yml index b942491a7..617cb686c 100644 --- a/mallinkSchedule/src/main/resources/application.yml +++ b/mallinkSchedule/src/main/resources/application.yml @@ -21,22 +21,22 @@ spring: cache-names: redis_cache #缓存的名字(可以不指定) redis: time-to-live: 60000ms #很重要,缓存的有效时间,以便缓存的过期(单位为毫秒) - rocketmq: - nameServer: 127.0.0.1:9876 - producer: - retry-times-when-send-async-failed: 0 - send-msg-timeout: 300000 - compress-msg-body-over-howmuch: 4096 - max-message-size: 4194304 - retry-another-broker-when-not-store-ok: false - retry-times-when-send-failed: 2 - rabbitmq: - host: 127.0.0.1 - port: 5672 - username: guest - password: guest - publisher-confirms: true - virtual-host: / +# rocketmq: +# nameServer: 127.0.0.1:9876 +# producer: +# retry-times-when-send-async-failed: 0 +# send-msg-timeout: 300000 +# compress-msg-body-over-howmuch: 4096 +# max-message-size: 4194304 +# retry-another-broker-when-not-store-ok: false +# retry-times-when-send-failed: 2 +# rabbitmq: +# host: 127.0.0.1 +# port: 5672 +# username: guest +# password: guest +# publisher-confirms: true +# virtual-host: / # MybatisPlus diff --git a/mallinkService/pom.xml b/mallinkService/pom.xml index 0ebc4d9a5..058791ccc 100644 --- a/mallinkService/pom.xml +++ b/mallinkService/pom.xml @@ -25,6 +25,13 @@ dozer 5.5.1 + + com.aliyun.openservices + ons-client + 1.8.5.Final + + + \ No newline at end of file diff --git a/mallinkService/src/main/java/com/iformall/mq/MQConfig.java b/mallinkService/src/main/java/com/iformall/mq/MQConfig.java index ffdc5daf1..4fb90461a 100644 --- a/mallinkService/src/main/java/com/iformall/mq/MQConfig.java +++ b/mallinkService/src/main/java/com/iformall/mq/MQConfig.java @@ -4,5 +4,6 @@ public class MQConfig { public static class Impl { public static final String RABBIT_MQ = "rabbitMQ"; public static final String ROCKET_MQ = "rocketMQ"; + public static final String ALIYUN_ROCKET_MQ = "aliyunRocketMQ"; } } diff --git a/mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java b/mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java new file mode 100644 index 000000000..fd3e4539d --- /dev/null +++ b/mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java @@ -0,0 +1,112 @@ +package com.iformall.mq.impl; + +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.UUID; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +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.iformall.domain.po.msg.BaseMsg; +import com.iformall.enums.EnumMsgRecordStatus; +import com.iformall.mapper.WxMsgRecordMapper; +import com.iformall.mq.MQConfig; +import com.iformall.mq.MqBaseProducer; +import com.iformall.service.WxMsgRecordService; +import com.iformall.utils.JsonUtil; + +/** + * @Auther: zfy 阿里云rocketmq + * @Date: 2019-01-31 + * @Description: RocketMQ消息生产者 + */ +@Service +@Profile(MQConfig.Impl.ALIYUN_ROCKET_MQ) +public class AliyunRocketMqMessageProducer implements MqBaseProducer { + private final Logger log = LoggerFactory.getLogger(this.getClass()); + + @Value("${spring.aliyunRocketmq.accessKeyId}") + private String accessKeyId; + + @Value("${spring.aliyunRocketmq.accessKeySecret}") + private String accessKeySecret; + + @Value("${spring.aliyunRocketmq.producerGroupId}") + private String producerGroupId; + + @Value("${spring.aliyunRocketmq.namesrvAddr}") + private String namesrvAddr; + + private static Map producers = new HashMap(); + + @Autowired + private WxMsgRecordMapper wxMsgRecordMapper; + + @Autowired + private WxMsgRecordService wxMsgRecordService; + + private Producer getProducer(String groupId) { + if (producers.containsKey(groupId)) { + return producers.get(groupId); + }else { + Properties properties = new Properties(); + // 您在控制台创建的 Group ID + properties.put(PropertyKeyConst.GROUP_ID, groupId); + // 鉴权用 AccessKeyId,在阿里云服务器管理控制台创建 + properties.put(PropertyKeyConst.AccessKey,accessKeyId); + // 鉴权用 AccessKeySecret,在阿里云服务器管理控制台创建 + properties.put(PropertyKeyConst.SecretKey, accessKeySecret); + // 设置 TCP 接入域名,进入控制台的实例详情页面,在页面上方选择实例后,在实例信息中的“获取接入点信息”区域查看 + properties.put(PropertyKeyConst.NAMESRV_ADDR,namesrvAddr); + + Producer producer = ONSFactory.createProducer(properties); + // 在发送消息前,必须调用 start 方法来启动 Producer,只需调用一次即可 + producer.start(); + producers.put(groupId, producer); + return producer; + } + } + + + + @Override + public void sendMessage(BaseMsg data, String topic, String tags, String keys) { + 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); + } + + } + +} diff --git a/mallinkWebSocketServer/src/main/resources/application-dev.yml b/mallinkWebSocketServer/src/main/resources/application-dev.yml index cd0c821ab..061b2518e 100644 --- a/mallinkWebSocketServer/src/main/resources/application-dev.yml +++ b/mallinkWebSocketServer/src/main/resources/application-dev.yml @@ -1,6 +1,7 @@ spring: profiles: - include: rabbitMQ + include: aliyunRocketMQ + #include: rabbitMQ # JDBC datasource: url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true @@ -71,7 +72,11 @@ spring: publisher-confirms: true publisher-returns: false virtual-host: / - + aliyunRocketmq: + accessKeyId:LTAI4G2hFwCBVhxCFbV7nBxf + accessKeySecret:LTAI4G2hFwCBVhxCFbV7nBxf + producerGroupId:GID_P_1 + namesrvAddr:http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 aws: clientRegion: cn-northwest-1 bucketName: iformall-net diff --git a/mallinkWebSocketServer/src/main/resources/application.yml b/mallinkWebSocketServer/src/main/resources/application.yml index d3aacff46..bdfa1b8d2 100644 --- a/mallinkWebSocketServer/src/main/resources/application.yml +++ b/mallinkWebSocketServer/src/main/resources/application.yml @@ -23,22 +23,22 @@ spring: cache-names: redis_cache #缓存的名字(可以不指定) redis: time-to-live: 60000ms #很重要,缓存的有效时间,以便缓存的过期(单位为毫秒) - rocketmq: - nameServer: 127.0.0.1:9876 - producer: - retry-times-when-send-async-failed: 0 - send-msg-timeout: 300000 - compress-msg-body-over-howmuch: 4096 - max-message-size: 4194304 - retry-another-broker-when-not-store-ok: false - retry-times-when-send-failed: 2 - rabbitmq: - host: 202.165.179.86 - port: 5672 - username: guest - password: guest - publisher-confirms: true - virtual-host: / +# rocketmq: +# nameServer: 127.0.0.1:9876 +# producer: +# retry-times-when-send-async-failed: 0 +# send-msg-timeout: 300000 +# compress-msg-body-over-howmuch: 4096 +# max-message-size: 4194304 +# retry-another-broker-when-not-store-ok: false +# retry-times-when-send-failed: 2 +# rabbitmq: +# host: 202.165.179.86 +# port: 5672 +# username: guest +# password: guest +# publisher-confirms: true +# virtual-host: / # MybatisPlus