RocketMQ——服务启动时初始化自定义RocketMQ的生产者Producer

深碍√TFBOYSˉ_ 2024-03-23 14:43 50阅读 0赞

## 需求 ##

1.  构造`RocketMQ`生产者。
2.  在服务启动时构造生产者，在服务关闭时销毁生产者。

## 代码模板 ##

### 抽象类 ###

@Slf4j
    public abstract class AbstractMqProducer {
        
    
        /**
         *  原生默认mq生产者
         */
        protected DefaultMQProducer producer;
    
        /**
         * 启动时，构造
         *
         * @throws MQClientException
         */
        @PostConstruct
        public abstract void start() throws MQClientException;
    
        /**
         * 关闭时，销毁
         */
        @PreDestroy
        public void shutdown() {
        
            log.info("AbstractMqProducer shutdown...");
            producer.shutdown();
        }
    
    }

### 生产者实现类 ###

@Component
    @Configuration
    @Slf4j
    @RequiredArgsConstructor
    public class DemoRocketMqProducer extends AbstractMqProducer {
        
    
        /**
         * 参数配置
         */
    	@Value("${rocketmq.producer.group}")
    	private String producerGroup;
    	@Value("${rocketmq.name-server}")
        private String nameServer;
    	@Value("${rocketmq.producer.send-message-timeout}")
        private String sendMessageTimeout;
       	@Value("${rocketmq.producer.compress-message-body-threshold}")
    	private String compressMessageBodyThreshold;
       	@Value("${rocketmq.producer.max-message-size}")
    	private String maxMessageSize;
       	@Value("${rocketmq.producer.retry-times-when-send-failed}")
    	private String retryTimesWhenSendFailed;
       	@Value("${rocketmq.producer.retry-times-when-send-async-failed}")
    	private String retryTimesWhenSendAsyncFailed;
       	@Value("${rocketmq.producer.retry-next-server}")
    	private String retryNextServer;
        /**
         * 启动producer
         *
         * @throws MQClientException
         */
        @Override
        public void start() throws MQClientException {
        
            //判空
            if (ObjectUtil.isNull(producer)) {
        
                log.info("init DemoRocketMqProducer begin... ...");
                //构建producer
                producer = new DefaultMQProducer(producerGroup);
                //name server地址
                producer.setNamesrvAddr(nameServer);
                //客户端实例名称
                producer.setInstanceName(UtilAll.getPid() + "#" + System.nanoTime());
                //发送消息超时时间，单位毫秒，默认3000
                producer.setSendMsgTimeout(sendMessageTimeout);
                //body超过多大开始压缩，单位字节，默认4096B
                producer.setCompressMsgBodyOverHowmuch(compressMessageBodyThreshold);
                //客户端限制的消息大小，单位字节，默认4194304
                producer.setMaxMessageSize(maxMessageSize);
                //如果消息发送失败，最大重试次数，默认2
                producer.setRetryTimesWhenSendFailed(retryTimesWhenSendFailed);
                //如果消息异步发送失败，最大重试次数，默认2
                producer.setRetryTimesWhenSendAsyncFailed(retryTimesWhenSendAsyncFailed);
                //如果发送消息返回sendResult，但sendStatus != SEND_OK，是否重试其他broker发送，默认false
                producer.setRetryAnotherBrokerWhenNotStoreOK(retryNextServer);
            }
            //启动producer
            producer.start();
            log.info("DemoRocketMqProducer is starting...");
        }
    
        /**
         * 发送消息
         *
         * @param topic
         * @param tag
         * @param key
         * @param dto
         * @param <T>
         * @return
         */
        public <T> boolean send(String topic, String tag, String key, T dto, String hashKey) {
        
            try {
        
                //转换dto为String类型
                String body = JSON.toJSONString(dto);
                //构建message
                Message message = new Message(topic, tag, key, body.getBytes("utf-8"));
                //发送消息
                SendResult sendResult = producer.send(message, new SelectMessageQueueByHash(), hashKey);
                log.info("DemoRocketMqProducer 消息生产结果：{}", sendResult);
                return true;
            } catch (Exception e) {
        
                //错误日志
                log.error("DemoRocketMqProducer 消息生产失败，错误信息：{}", e);
                //异常处理
                throw new RuntimeException(e.getMessage());
            }
        }
    
        /**
         * 关闭producer
         */
        @Override
        @PreDestroy
        public void shutdown() {
        
            //关闭producer
            producer.shutdown();
            log.warn("DemoRocketMqProducer shutdown...");
        }
    }

### 注解详解 ###

#### @PostConstruct ####

##### 介绍 #####

1.  `@PostConstruct`注解被用来修饰一个非静态的`void()`方法。
2.  被 `@PostConstruct`注解修饰的方法会在服务加载`Servlet`时运行，只会被执行一次。
3.  `@PostConstruct`注解在`init()`方法之前、构造函数之后执行。
4.  上下文启动顺序：`服务加载Servlet->Constructor(构造方法) -> @Autowired(依赖注入) -> @PostConstruct(注解修饰的方法)->init()方法`

##### 使用场景 #####

若想在生成对象时完成某些初始化操作，此时这些初始化需要依赖于一些依赖注入，则无法在构造函数中实现。此时，可使用`@PostConstruct`注解修饰一个方法来完成初始化，`@PostConstruct`注解修饰的方法将会在依赖注入完成后被自动调用。

##### 源码 #####

@Documented
    @Retention (RUNTIME)
    @Target(METHOD)
    public @interface PostConstruct {
        
    }

#### @PreDestroy ####

##### 介绍 #####

1.  `@PreDestroy`注解被用来修饰一个非静态的`void()`方法。
2.  被 `@PreDestroy`注解修饰的方法会在服务卸载`Servlet`时运行，只会被执行一次。
3.  `@PreDestroy`注解在`Servlet`卸载之前，`destroy()`方法之后执行。
4.  上下文停止顺序：`destroy()方法 -> @PreDestroy(注解修饰的方法)-> bean销毁->服务卸载Servlet`

##### 源码 #####

@Documented
    @Retention (RUNTIME)
    @Target(METHOD)
    public @interface PreDestroy {
        
    }

### 配置文件详解 ###

# rocketmq消息队列
    rocketmq:
      # 指定 nameServer
      name-server: 127.0.0.1:9876;127.0.0.1:9877
      # Producer 生产者
      producer:
      	# 生产组
        group: my-group
        # 发送消息超时时间，单位毫秒，默认3000
        send-message-timeout: 3000
        # body超过多大开始压缩，单位字节，默认4096B
        compress-message-body-threshold: 4096
        # 客户端限制的消息大小，单位字节，默认4194304
        max-message-size: 4194304
        # 如果消息发送失败，最大重试次数，默认2
        retry-times-when-send-failed: 2
        # 如果消息异步发送失败，最大重试次数，默认2
        retry-times-when-send-async-failed: 2
        # 如果发送消息返回sendResult，但sendStatus != SEND_OK，是否重试其他broker发送，默认false
        retry-next-server: false