RocketMqProducer 消息生产者定义范例

小咪咪 2022-10-28 03:54 149阅读 0赞

DefaultMQProducer参数了解：  
https://blog.csdn.net/jiangshubian/article/details/100054281

自定义选择器(可不定义)

@Component
    public class GenericQueueSelector implements MessageQueueSelector { 
        @Override
        public MessageQueue select(List<MessageQueue> list, Message message, Object o) { 
            long id = NumberUtils.toLong(String.valueOf(o));
            int index = (int) (id % list.size());
            return list.get(index);
        }
    }

定义消息生产者

@Slf4j
    @Component
    public class RocketMqProducer implements FactoryBean<DefaultMQProducer>, InitializingBean, DisposableBean { 
    
        @Resource(name = "messageSenderThreadPool")
        private ThreadPoolTaskExecutor messageSenderThreadPool;
        @Autowired
        private GenericQueueSelector selector;
    
        @Value("${rocketmq.producer.producerGroupName}")
        private String producerGroupName;
        @Value("${rocketmq.producer.instanceName}")
        private String instanceName;
        @Value("${rocketmq.producer.nameServerAddr}")
        private String nameServerAddr;
        @Value("${rocketmq.producer.defaultTopicQueueNums}")
        private int defaultTopicQueueNums;
        @Value("${rocketmq.producer.sendMsgTimeout}")
        private int sendMsgTimeout;
        @Value("${rocketmq.producer.retryTimesWhenSendFailed}")
        private int retryTimesWhenSendFailed;
    
        private DefaultMQProducer defaultMQProducer;
    
        @Override
        public void destroy() throws Exception { 
            if (null != defaultMQProducer) { 
                defaultMQProducer.shutdown();
            }
        }
    
        @Override
        public DefaultMQProducer getObject() throws Exception { 
            return this.defaultMQProducer;
        }
    
        @Override
        public Class<?> getObjectType() { 
            return DefaultMQProducer.class;
        }
    
        @Override
        public void afterPropertiesSet() throws Exception { 
            defaultMQProducer = new DefaultMQProducer(producerGroupName);
            defaultMQProducer.setInstanceName(instanceName);
            defaultMQProducer.setNamesrvAddr(nameServerAddr);
            defaultMQProducer.setDefaultTopicQueueNums(defaultTopicQueueNums);
            defaultMQProducer.setSendMsgTimeout(sendMsgTimeout);
            defaultMQProducer.setRetryTimesWhenSendFailed(retryTimesWhenSendFailed);
            defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(true);
            defaultMQProducer.start();
            log.info("rocketMQ producer group [{}] started", producerGroupName);
        }
    
    
        public ListenableFuture<SendResult> sendTest(String topic, String tags, String keys, String body) { 
            ListenableFuture<SendResult> future = messageSenderThreadPool.submitListenable(() -> { 
                try { 
                    log.info("topic {}, tags:{}, keys: {}", topic, tags, keys);
                    return this.defaultMQProducer.send(new Message(topic, tags, keys, body.getBytes()), selector, keys);
                } catch (Exception e) { 
                    log.error("sendMessageError: {}", e.getMessage(), e);
                }
                return new SendResult();
            });
    
            return future;
        }
    }

使用  
先顶一个一个service，然后进行注入

@Autowired
        private RocketMqProducer rocketMqProducer;
    
    	@SneakyThrows
        @Override
        public boolean sendMQTest(InvoiceApplyBO invoiceApplyBO) { 
    
           String yourKes = "";
        	String yourMsg = "";
            
            ListenableFuture<SendResult> future = rocketMqProducer.sendOrderly(producerTopicAndTags, producerTopicAndTags, yourKes, yourMsg);
            SendResult sendResult = future.get();
            if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) { 
                return true;
            } else { 
                log.warn("sendOrderly failed, sendResult -> {}", sendResult);
            }
    
            return false;
        }