| @@ -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 | |||
| @@ -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 | |||
| @@ -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 | |||
| @@ -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 | |||
| @@ -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<Subscription, MessageListener> subscriptionTable = new HashMap<Subscription, MessageListener>(); | |||
| 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(); | |||
| @@ -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" | |||
| @@ -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 | |||
| @@ -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 | |||
| @@ -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()); | |||