Просмотр исходного кода

接入阿里云rocketmq

release_toaliyun_real
xiaohanzi 6 лет назад
Родитель
Сommit
1fe96beb84
19 измененных файлов: 345 добавлений и 113 удалений
  1. +7
    -1
      mallinkAdmin/src/main/resources/application-dev.yml
  2. +16
    -16
      mallinkAdmin/src/main/resources/application.yml
  3. +7
    -2
      mallinkBApi/src/main/resources/application-dev.yml
  4. +9
    -9
      mallinkBApi/src/main/resources/application.yml
  5. +7
    -2
      mallinkCApi/src/main/resources/application-dev.yml
  6. +9
    -9
      mallinkCApi/src/main/resources/application.yml
  7. +7
    -2
      mallinkCallback/src/main/resources/application-dev.yml
  8. +16
    -16
      mallinkCallback/src/main/resources/application.yml
  9. +71
    -0
      mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java
  10. +7
    -2
      mallinkMQConsumer/src/main/resources/application-dev.yml
  11. +16
    -16
      mallinkMQConsumer/src/main/resources/application.yml
  12. +7
    -2
      mallinkPosApi/src/main/resources/application-dev.yml
  13. +7
    -2
      mallinkSchedule/src/main/resources/application-dev.yml
  14. +16
    -16
      mallinkSchedule/src/main/resources/application.yml
  15. +7
    -0
      mallinkService/pom.xml
  16. +1
    -0
      mallinkService/src/main/java/com/iformall/mq/MQConfig.java
  17. +112
    -0
      mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java
  18. +7
    -2
      mallinkWebSocketServer/src/main/resources/application-dev.yml
  19. +16
    -16
      mallinkWebSocketServer/src/main/resources/application.yml

+ 7
- 1
mallinkAdmin/src/main/resources/application-dev.yml Просмотреть файл

@@ -1,6 +1,7 @@
spring: spring:
profiles: profiles:
include: rabbitMQ
include: aliyunRocketMQ
#include: rabbitMQ
# JDBC # JDBC
datasource: 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 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-confirms: true
publisher-returns: false publisher-returns: false
virtual-host: / 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: flyway:
enabled: false enabled: false




+ 16
- 16
mallinkAdmin/src/main/resources/application.yml Просмотреть файл

@@ -21,22 +21,22 @@ spring:
cache-names: redis_cache #缓存的名字(可以不指定) cache-names: redis_cache #缓存的名字(可以不指定)
redis: redis:
time-to-live: 60000ms #很重要,缓存的有效时间,以便缓存的过期(单位为毫秒) 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
flowable: flowable:
async-executor-activate: false async-executor-activate: false


+ 7
- 2
mallinkBApi/src/main/resources/application-dev.yml Просмотреть файл

@@ -1,6 +1,7 @@
spring: spring:
profiles: profiles:
include: rabbitMQ
include: aliyunRocketMQ
#include: rabbitMQ
# JDBC # JDBC
datasource: datasource:
url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true 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-confirms: true
publisher-returns: false publisher-returns: false
virtual-host: / 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: aws:
clientRegion: cn-northwest-1 clientRegion: cn-northwest-1
bucketName: iformall-net bucketName: iformall-net


+ 9
- 9
mallinkBApi/src/main/resources/application.yml Просмотреть файл

@@ -12,15 +12,15 @@ spring:
date-format: yyyy-MM-dd HH:mm:ss date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+8 time-zone: GMT+8
default-property-inclusion: non_null 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 # MybatisPlus
mybatis-plus: mybatis-plus:


+ 7
- 2
mallinkCApi/src/main/resources/application-dev.yml Просмотреть файл

@@ -1,6 +1,7 @@
spring: spring:
profiles: profiles:
include: rabbitMQ
include: aliyunRocketMQ
#include: rabbitMQ
# JDBC # JDBC
datasource: datasource:
url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true 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-confirms: true
publisher-returns: false publisher-returns: false
virtual-host: / 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: aws:
clientRegion: cn-northwest-1 clientRegion: cn-northwest-1
bucketName: iformall-net bucketName: iformall-net


+ 9
- 9
mallinkCApi/src/main/resources/application.yml Просмотреть файл

@@ -12,15 +12,15 @@ spring:
date-format: yyyy-MM-dd HH:mm:ss date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+8 time-zone: GMT+8
default-property-inclusion: non_null 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 # MybatisPlus
mybatis-plus: mybatis-plus:


+ 7
- 2
mallinkCallback/src/main/resources/application-dev.yml Просмотреть файл

@@ -1,6 +1,7 @@
spring: spring:
profiles: profiles:
include: rabbitMQ
include: aliyunRocketMQ
#include: rabbitMQ
# JDBC # JDBC
datasource: datasource:
url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true 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-confirms: true
publisher-returns: false publisher-returns: false
virtual-host: / 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: aws:
clientRegion: cn-northwest-1 clientRegion: cn-northwest-1
bucketName: iformall-net bucketName: iformall-net


+ 16
- 16
mallinkCallback/src/main/resources/application.yml Просмотреть файл

@@ -21,22 +21,22 @@ spring:
cache-names: redis_cache #缓存的名字(可以不指定) cache-names: redis_cache #缓存的名字(可以不指定)
redis: redis:
time-to-live: 60000ms #很重要,缓存的有效时间,以便缓存的过期(单位为毫秒) 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: aop:
proxy-target-class: true proxy-target-class: true
auto: true auto: true


+ 71
- 0
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);
// }
}

+ 7
- 2
mallinkMQConsumer/src/main/resources/application-dev.yml Просмотреть файл

@@ -1,6 +1,7 @@
spring: spring:
profiles: profiles:
include: rabbitMQ
include: aliyunRocketMQ
#include: rabbitMQ
# JDBC # JDBC
datasource: datasource:
url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true 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-confirms: true
publisher-returns: false publisher-returns: false
virtual-host: / virtual-host: /

aliyunRocketmq:
accessKeyId:LTAI4G2hFwCBVhxCFbV7nBxf
accessKeySecret:IIRfygUsnNNP1KaF3OTRdo0z9KzkbF
consumerGroupId:GID_C_1
namesrvAddr://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080
aws: aws:
clientRegion: cn-northwest-1 clientRegion: cn-northwest-1
bucketName: iformall-net bucketName: iformall-net


+ 16
- 16
mallinkMQConsumer/src/main/resources/application.yml Просмотреть файл

@@ -21,22 +21,22 @@ spring:
cache-names: redis_cache #缓存的名字(可以不指定) cache-names: redis_cache #缓存的名字(可以不指定)
redis: redis:
time-to-live: 60000ms #很重要,缓存的有效时间,以便缓存的过期(单位为毫秒) 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 # MybatisPlus


+ 7
- 2
mallinkPosApi/src/main/resources/application-dev.yml Просмотреть файл

@@ -1,6 +1,7 @@
spring: spring:
profiles: profiles:
include: rabbitMQ
include: aliyunRocketMQ
#include: rabbitMQ
# JDBC # JDBC
datasource: datasource:
url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true 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-confirms: true
publisher-returns: false publisher-returns: false
virtual-host: / 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: aws:
clientRegion: cn-northwest-1 clientRegion: cn-northwest-1
bucketName: iformall-net bucketName: iformall-net


+ 7
- 2
mallinkSchedule/src/main/resources/application-dev.yml Просмотреть файл

@@ -1,6 +1,7 @@
spring: spring:
profiles: profiles:
include: rabbitMQ
include: aliyunRocketMQ
#include: rabbitMQ
# JDBC # JDBC
datasource: datasource:
url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true 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-confirms: true
publisher-returns: false publisher-returns: false
virtual-host: / 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: aws:
clientRegion: cn-northwest-1 clientRegion: cn-northwest-1
bucketName: iformall-net bucketName: iformall-net


+ 16
- 16
mallinkSchedule/src/main/resources/application.yml Просмотреть файл

@@ -21,22 +21,22 @@ spring:
cache-names: redis_cache #缓存的名字(可以不指定) cache-names: redis_cache #缓存的名字(可以不指定)
redis: redis:
time-to-live: 60000ms #很重要,缓存的有效时间,以便缓存的过期(单位为毫秒) 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 # MybatisPlus


+ 7
- 0
mallinkService/pom.xml Просмотреть файл

@@ -25,6 +25,13 @@
<artifactId>dozer</artifactId> <artifactId>dozer</artifactId>
<version>5.5.1</version> <version>5.5.1</version>
</dependency> </dependency>
<dependency>
<groupId>com.aliyun.openservices</groupId>
<artifactId>ons-client</artifactId>
<version>1.8.5.Final</version>
</dependency>
</dependencies> </dependencies>


</project> </project>

+ 1
- 0
mallinkService/src/main/java/com/iformall/mq/MQConfig.java Просмотреть файл

@@ -4,5 +4,6 @@ public class MQConfig {
public static class Impl { public static class Impl {
public static final String RABBIT_MQ = "rabbitMQ"; public static final String RABBIT_MQ = "rabbitMQ";
public static final String ROCKET_MQ = "rocketMQ"; public static final String ROCKET_MQ = "rocketMQ";
public static final String ALIYUN_ROCKET_MQ = "aliyunRocketMQ";
} }
} }

+ 112
- 0
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<String,Producer> producers = new HashMap<String,Producer>();

@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);
}

}

}

+ 7
- 2
mallinkWebSocketServer/src/main/resources/application-dev.yml Просмотреть файл

@@ -1,6 +1,7 @@
spring: spring:
profiles: profiles:
include: rabbitMQ
include: aliyunRocketMQ
#include: rabbitMQ
# JDBC # JDBC
datasource: datasource:
url: jdbc:mysql://rm-2zel9i9t555zy7lft.mysql.rds.aliyuncs.com:3306/mallink?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useUnicode=true&useSSL=false&useAffectedRows=true 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-confirms: true
publisher-returns: false publisher-returns: false
virtual-host: / 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: aws:
clientRegion: cn-northwest-1 clientRegion: cn-northwest-1
bucketName: iformall-net bucketName: iformall-net


+ 16
- 16
mallinkWebSocketServer/src/main/resources/application.yml Просмотреть файл

@@ -23,22 +23,22 @@ spring:
cache-names: redis_cache #缓存的名字(可以不指定) cache-names: redis_cache #缓存的名字(可以不指定)
redis: redis:
time-to-live: 60000ms #很重要,缓存的有效时间,以便缓存的过期(单位为毫秒) 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 # MybatisPlus


Загрузка…
Отмена
Сохранить