devbox
很楠不爱3
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Java基于rocketmq的订单生产与消费场景示例_java rocketmq的生产与消费

作者：很楠不爱3 | 2024-04-25 11:14:15

赞

踩

java rocketmq的生产与消费

一、依赖配置

1、maven pom.xml引入依赖

  1. <properties>
  2.         <rocketmq-client.version>4.5.2</rocketmq-client.version>
  3. </properties>
  4. <dependency>
  5.          <groupId>org.apache.rocketmq</groupId>
  6.          <artifactId>rocketmq-client</artifactId>
  7.          <version>${rocketmq-client.version}</version>
  8. </dependency>

2、生产者application.yml加上配置

  1. rocketmq:
  2.   producer:
  3.     isOnOff: on
  4.     groupName: k12OrderNotifyGroup
  5.     namesrvAddr: rocket-01:9876;rocket-02:9876;rocket-03:9876
  6.     maxMessageSize: 4096
  7.     sendMsgTimeout: 3000
  8.     retryTimesWhenSendFailed: 2
  9.     topic: k12_order

 3、消费者application.yml加上配置

  1. rocketmq:
  2.   consumer:
  3.     #该应用是否启用消费者
  4.     isOnOff: on
  5.     groupName: k12OrderNotifyGroup
  6.     #mq的nameserver地址
  7.     namesrvAddr: rocket-01:9876;rocket-02:9876;rocket-03:9876
  8.     #该消费者订阅的主题和tags("*"号表示订阅该主题下所有的tags),格式：topic~tag1||tag2||tag3;topic2~*;
  9.     topics: k12_order~*;
  10.     consumeThreadMin: 20
  11.     consumeThreadMax: 64
  12.     #设置一次消费消息的条数，默认为1条
  13.     consumeMessageBatchMaxSize: 1
  14.     reconsumeTimes: 3
  15.     topic: k12_order

二、Producer服务生产订单

1、生产启动配置类

  1. import lombok.Data;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.apache.rocketmq.client.exception.MQClientException;
  4. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  5. import org.springframework.beans.factory.annotation.Value;
  6. import org.springframework.boot.context.properties.ConfigurationProperties;
  7. import org.springframework.context.annotation.Bean;
  8. import org.springframework.context.annotation.Configuration;
  9. import org.springframework.util.ObjectUtils;
  10.  
  11. import java.util.UUID;
  12.  
  13. @Slf4j
  14. @Data
  15. @ConfigurationProperties(prefix = "rocketmq-producer")
  16. public class RocketMQProducerConfig {
  17.  
  18.     /**
  19.      * 生产/消费组
  20.      */
  21.     private String groupName;
  22.  
  23.     /**
  24.      * 地址
  25.      */
  26.     private String namesrvAddr;
  27.  
  28.     /**
  29.      * 最大生产量
  30.      */
  31.     private Integer maxMessageSize;
  32.  
  33.     /**
  34.      * 往队列中发送消息超时时间
  35.      */
  36.     private Integer sendMsgTimeout;
  37.  
  38.     /**
  39.      * 超时失败重发次数
  40.      */
  41.     private Integer retryTimesWhenSendFailed;
  42.  
  43.     /**
  44.      * topic主题
  45.      */
  46.     private String topic;
  47.  
  48.     @Bean
  49.     public DefaultMQProducer getRocketMQProducer() {
  50.         DefaultMQProducer defaultMQProducer = new DefaultMQProducer(this.groupName);
  51.         defaultMQProducer.setNamesrvAddr(this.namesrvAddr);
  52.  
  53.         //如果需要同一个jvm中不同的producer往不同的mq集群发送消息，需要设置不同的instanceName
  54.         //defaultMQProducer.setInstanceName(UUID.randomUUID().toString());
  55.  
  56.         if (!ObjectUtils.isEmpty(this.maxMessageSize)) {
  57.             defaultMQProducer.setMaxMessageSize(this.maxMessageSize);
  58.         }
  59.         if (!ObjectUtils.isEmpty(this.sendMsgTimeout)) {
  60.             defaultMQProducer.setSendMsgTimeout(this.sendMsgTimeout);
  61.         }
  62.         if (!ObjectUtils.isEmpty(this.retryTimesWhenSendFailed)) {
  63.             defaultMQProducer.setRetryTimesWhenSendFailed(this.retryTimesWhenSendFailed);
  64.         }
  65.  
  66.         try {
  67.             //生产者启动
  68.             defaultMQProducer.start();
  69.             log.info(String.format(
  70.                     "producer start, groupName:[%s],namesrvAddr:[%s]"
  71.                     , this.groupName
  72.                     , this.namesrvAddr
  73.                     )
  74.             );
  75.         } catch (MQClientException e) {
  76.             log.error(String.format(
  77.                     "producer is error, {}"
  78.                     , e.getMessage()
  79.                     , e
  80.                     )
  81.             );
  82.         }
  83.         return defaultMQProducer;
  84.     }
  85. }

2、生产者类

接口

  1. public interface Producer {
  2.  
  3.     void send(String topic, String tags, String keys, String body);
  4.  
  5.     void sendDelay(String topic, String tags, String keys, String body, Integer delayTimeLevel);
  6. }

 实现类

  1. import com.xiyunerp.yzbizcenter.k12.order.dmm.Producer;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.apache.rocketmq.client.exception.MQBrokerException;
  4. import org.apache.rocketmq.client.exception.MQClientException;
  5. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  6. import org.apache.rocketmq.client.producer.SendResult;
  7. import org.apache.rocketmq.common.message.Message;
  8. import org.apache.rocketmq.remoting.exception.RemotingException;
  9. import org.springframework.stereotype.Component;
  10.  
  11. import javax.annotation.Resource;
  12.  
  13. @Slf4j
  14. @Component
  15. public class RocketMQProducer implements Producer {
  16.  
  17.     @Resource
  18.     private DefaultMQProducer defaultMQProducer;
  19.  
  20.     @Override
  21.     public void send(String topic, String tags, String keys, String body) {
  22.         Message message = new Message(topic, tags, keys, body.getBytes());
  23.         log.debug("消息发送信息内容：{}", body);
  24.         SendResult sendResult;
  25.         try {
  26.             sendResult = defaultMQProducer.send(message);
  27.             log.debug("消息发送响应信息: {}", sendResult.toString());
  28.         } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
  29.             log.debug("发送消息异常信息: {}", e.getMessage());
  30.         }
  31.     }
  32.  
  33.     @Override
  34.     public void sendDelay(String topic, String tags, String keys, String body, Integer delayTimeLevel) {
  35.         Message message = new Message(topic, tags, keys, body.getBytes());
  36.         message.setDelayTimeLevel(delayTimeLevel);
  37.         log.debug("延迟消息发送信息内容：{}", body);
  38.         SendResult sendResult;
  39.         try {
  40.             sendResult = defaultMQProducer.send(message);
  41.             log.debug("延迟消息发送响应信息: {}", sendResult.toString());
  42.         } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
  43.             log.debug("延迟发送消息异常信息: {}", e.getMessage());
  44.         }
  45.     }
  46. }

3、业务类（调用生产者）

  1. import com.xiyunerp.yzbizcenter.k12.order.config.RocketMQConfig;
  2. import com.xiyunerp.yzbizcenter.k12.order.zys.dmm.RocketMQProducer;
  3. import com.xiyunerp.yzbizcenter.k12.utility.context.BeanUtils;
  4. import com.xiyunerp.yzbizcenter.k12.utility.utils.JsonUtils;
  5. import com.xiyunerp.yzbizcenter.k12.utility.utils.StringUtils;
  6. import lombok.extern.slf4j.Slf4j;
  7.  
  8. import org.springframework.stereotype.Service;
  9.  
  10. import java.util.List;
  11.  
  12. @Slf4j
  13. @Service
  14. public class OrderProducerService {
  15.     
  16.     private RocketMQProducerConfig  rocketMQProducerConfig  = 
  17.             BeanUtils.getBean("rocketMQProducerConfig ", RocketMQProducerConfig .class);
  18.  
  19.     private RocketMQProducer rocketMQProducer = BeanUtils.getBean("rocketMQProducer", RocketMQProducer.class);
  20.  
  21.     public void sendDelay() {
  22.         //处理自己其他的业务逻辑,设置相关参数
  23.         
  24.         //调用生产类进行往队列中发送消息
  25.         String topic = "";
  26.         String tags = "";
  27.         String keys = "";
  28.         String body = "";
  29.         Integer delayTimeLevel = 4;
  30.         try {
  31.             //指定延迟时间
  32.             rocketMQProducer.sendDelay(topic, tags, keys, body, delayTimeLevel);
  33.             //rocketMQProducer.send(topic, tags, keys, body);
  34.         } catch (Exception e) {
  35.             log.debug("rocketMQ producer error, {}", e.getMessage());
  36.         }
  37.     }
  38. }

三、Consumer服务生产订单

1、消费启动配置类

  1. import lombok.Data;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  4. import org.apache.rocketmq.client.exception.MQClientException;
  5. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  6. import org.springframework.beans.factory.annotation.Value;
  7. import org.springframework.context.annotation.Bean;
  8. import org.springframework.context.annotation.Configuration;
  9.  
  10. import javax.annotation.Resource;
  11. import java.util.UUID;
  12.  
  13. @Slf4j
  14. @Data
  15. @ConfigurationProperties(prefix = "rocketmq-consumer")
  16. public class RocketMQConsumerConfig {
  17.  
  18.     private String namesrvAddr;
  19.  
  20.     private String groupName;
  21.  
  22.     private Integer consumeThreadMin;
  23.  
  24.     private Integer consumeThreadMax;
  25.  
  26.     private String topics;
  27.  
  28.     private Integer consumeMessageBatchMaxSize;
  29.  
  30.     private Integer reconsumeTimes;
  31.  
  32.     private String topic;
  33.  
  34.     @Resource
  35.     private RocketListenerProcessor mqMessageListenerProcessor;
  36.  
  37.     @Bean
  38.     public DefaultMQPushConsumer getRocketMQConsumer() {
  39.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
  40.         consumer.setNamesrvAddr(namesrvAddr);
  41.         consumer.setInstanceName(UUID.randomUUID().toString());
  42.         consumer.setConsumeThreadMin(consumeThreadMin);
  43.         consumer.setConsumeThreadMax(consumeThreadMax);
  44.         consumer.registerMessageListener(mqMessageListenerProcessor);
  45.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
  46.         /**
  47.          * 设置消费模型，集群还是广播，默认为集群
  48.          */
  49.         //consumer.setMessageModel(MessageModel.CLUSTERING);
  50.         consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
  51.         try {
  52.             String[] topicTagsArr = topics.split(";");
  53.             for (String topicTags : topicTagsArr) {
  54.                 String[] topicTag = topicTags.split("~");
  55.                 consumer.subscribe(topicTag[0], topicTag[1]);
  56.             }
  57.             consumer.start();
  58.             log.debug("consumer is start, groupName:{}, topics:{}, namesrvAddr:{}", groupName, topics, namesrvAddr);
  59.         } catch (MQClientException e) {
  60.             log.error("consumer is start, groupName:{}, topics:{}, namesrvAddr:{}", groupName, topics, namesrvAddr, e);
  61.         }
  62.         return consumer;
  63.     }
  64. }

2、消费者类（调用业务类处理业务）

  1. import com.xiyunerp.yzbizcenter.k12.mini.alipay.service.trade.AliPayOrderSyncService;
  2. import com.xiyunerp.yzbizcenter.k12.utility.utils.CollectionUtils;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
  5. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  6. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  7. import org.apache.rocketmq.common.message.MessageExt;
  8. import org.springframework.data.redis.core.RedisTemplate;
  9. import org.springframework.stereotype.Component;
  10.  
  11. import javax.annotation.Resource;
  12. import java.util.List;
  13. import java.util.concurrent.TimeUnit;
  14.  
  15. @Slf4j
  16. @Component
  17. public class RocketListenerProcessor implements MessageListenerConcurrently {
  18.  
  19.     @Resource
  20.     RocketMQConsumerConfig rocketMQConfig;
  21.  
  22.     @Resource
  23.     AliPayOrderSyncService aliPayOrderSyncService;//处理业务类，自定义
  24.  
  25.     @Resource
  26.     RedisTemplate redisTemplate;
  27.  
  28.     private final static Long EXPIRE_TIME_MILLI = 1 * 60 * 1000L;
  29.  
  30.     private final static Long INCREMENT_STEP = 1L;
  31.  
  32.     @Override
  33.     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
  34.         if (CollectionUtils.isEmpty(list)) {
  35.             log.debug("接受到的消息为空，不处理，直接返回成功");
  36.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  37.         }
  38.         for (MessageExt messageExt : list) {
  39.             if (this.barrier(messageExt.getMsgId(), EXPIRE_TIME_MILLI)) {
  40.                 continue;
  41.             }
  42.             if (messageExt.getTopic().equalsIgnoreCase(rocketMQConfig.getTopic())) {
  43.                 int reconsume = messageExt.getReconsumeTimes();
  44.                 if (reconsume == rocketMQConfig.getReconsumeTimes()) {
  45.                     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  46.                 }
  47.  
  48.                 log.debug("接受到的消息为：{}", messageExt.toString());
  49.                 log.debug("接受到的消息为：{}, {}", messageExt.getTags(), Long.parseLong(new String(messageExt.getBody())));
  50.                 //调用业务类消费数据,自定义
  51.                 //aliPayOrderSyncService.sendOrder(messageExt);
  52.             }
  53.         }
  54.         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  55.     }
  56.     /**
  57.      * 做幂等避免重复消费
  58.      * @param key
  59.      * @param expireMillis
  60.      * @return
  61.      */
  62.     public Boolean barrier(String key, Long expireMillis) {
  63.         Long count = this.increment(key, expireMillis);
  64.         if (count > 1) {
  65.             return Boolean.TRUE;
  66.         }
  67.         return Boolean.FALSE;
  68.     }
  69.  
  70.     public Long increment(String key, Long expireMillis) {
  71.         Long count = redisTemplate.opsForValue().increment(key, INCREMENT_STEP);
  72.         if (1 == count) {
  73.             redisTemplate.expire(key, expireMillis, TimeUnit.MILLISECONDS);
  74.         }
  75.         return count;
  76.     }
  77. }

3、业务类（被消费类调用，处理自己业务）

  1. @Service
  2. public class AliPayOrderSyncService {
  3. }

 

声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号