RocketMQ

public static void producer() throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
    DefaultMQProducer producer = new DefaultMQProducer("group1");
    producer.setNamesrvAddr("127.0.0.1:9876");//设置命名服务器地址
    producer.start();
    Message message = new Message("topic1", "hello world!".getBytes(StandardCharsets.UTF_8));//参数为topic和要发送的消息
    SendResult result = producer.send(message);
    System.out.println("result = " + result);
    producer.shutdown();
}

public static void consumer() throws MQClientException {
    DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer("group1");
    pushConsumer.setNamesrvAddr("127.0.0.1:9876");
    pushConsumer.subscribe("topic1","*");//订阅一个消息，"*"匹配规则
    pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
            list.forEach(messageExt -> {
                System.out.println("messageExt = " + messageExt);
                System.out.println("body = " + new String(messageExt.getBody()));
            });
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    pushConsumer.start();
}

producer.send(message, new SendCallback() {
    @Override
    public void onSuccess(SendResult sendResult) {
        System.out.println("发送成功的结果 = " + sendResult);
    }

    @Override
    public void onException(Throwable throwable) {
        System.out.println("发送失败的异常 = " + throwable);
    }
});

@RestController
@RequestMapping("/producer")
public class ProducerController {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @GetMapping("/send")
    public String sendMsg(){
        rocketMQTemplate.convertAndSend("topic1:tag1","小石潭记");
        SendResult sendResult = rocketMQTemplate.syncSend("topic1:tag1", MessageBuilder.withPayload("同步消息").build());
        System.out.println("sendResult = " + sendResult);
        return "Ok!";
    }
}

@Service
@RocketMQMessageListener(topic = "topic1",consumerGroup = "group1",selectorExpression = "tag1")
public class ConsumerServer implements RocketMQListener<String> {
    @Override
    public void onMessage(String s) {
        System.out.println("消息 = " + s);
    }
}

public static void producerTransaction() throws MQClientException{
    TransactionMQProducer producer = new TransactionMQProducer("group1");
    producer.setNamesrvAddr("127.0.0.1:9876");//设置命名服务器地址
    producer.setTransactionListener(new TransactionListener() {
        @Override
        public LocalTransactionState executeLocalTransaction(Message message, Object o) {
            //模拟
            String isOk = "OK";
            LocalTransactionState state = null;
            if(isOk.equals("OK")){
                //正常事务执行
                state = LocalTransactionState.COMMIT_MESSAGE;//返回值为此会通知消息服务器将消息提交到消息队列当中
            }else if(isOk.equals("NO")){
                //事务异常执行
                state = LocalTransactionState.ROLLBACK_MESSAGE;//会通知消息服务器让其删除消息
            }else {
                state = LocalTransactionState.UNKNOW;//会触发事务补偿
            }
            return state;
        }

        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
            //补偿事务过程
            System.out.println(" 事务补偿 ");
            //根据检测结果返回LocalTransactionState.COMMIT_MESSAGE、LocalTransactionState.ROLLBACK_MESSAGE、LocalTransactionState.UNKNOW 当也为UNKNOW时则人工介入。
            return LocalTransactionState.COMMIT_MESSAGE;
        }
    });
    producer.start();
    Message message = new Message("topic1","tag1", "hello world!".getBytes(StandardCharsets.UTF_8));//参数为topic和要发送的消息
    SendResult result = producer.sendMessageInTransaction(message,null);
    System.out.println("result = " + result);
    //发送事务消息不能关闭生产者
}