1、同步消息
即时性较强，重要的消息，且必须有回执的消息，例如短信，通知（转账成功）
生产者：
public class Producer {
public static void main(String[] args) throws Exception{
DefaultMQProducer producer=new DefaultMQProducer("group1");
producer.setNamesrvAddr("192.168.23.127:9876");
producer.start();
for (int i = 1; i <= 5; i++) {
Message msg = new Message("topic2",("同步消息：hello rocketmq "+i).getBytes("UTF-8"));
//同步消息发送
SendResult result = producer.send(msg);
System.out.println("返回结果："+result);
}
producer.shutdown();
}
}
消费者：
public class Consumer {
public static void main(String[] args) throws Exception{
DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("group1");
consumer.setNamesrvAddr("192.168.23.127:9876");
consumer.subscribe("topic2","*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt msg : list) {
//System.out.println("收到消息："+msg);
System.out.println("消息："+new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功处理, mq 收到这个 标记后相同的消息讲不会再次发给消费者
}
});
consumer.start();// 开启多线程 监控消息,持续运行
System.out.println("接收消息服务已运行");
}
}测试：
2、异步消息
即时性较弱，但需要有回执的消息，例如订单中的某些信息
生产者：
public class Producer {
public static void main(String[] args) throws Exception{
DefaultMQProducer producer=new DefaultMQProducer("group1");
producer.setNamesrvAddr("192.168.23.127:9876");
producer.start();
for (int i = 1; i <= 5; i++) {
//异步消息发送
Message msg = new Message("topic2",("异步消息：hello rocketmq "+i).getBytes("UTF-8"));
producer.send(msg, new SendCallback() {
//表示成功返回结果
@Override
public void onSuccess(SendResult sendResult) {
System.out.println(sendResult);
}
//表示发送消息失败
@Override
public void onException(Throwable throwable) {
System.out.println(throwable);
}
});
}
//添加一个休眠操作，确保异步消息返回后能够输出
TimeUnit.SECONDS.sleep(10);
producer.shutdown();
}
}
消费者：
public class Consumer {
public static void main(String[] args) throws Exception{
DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("group1");
consumer.setNamesrvAddr("192.168.23.127:9876");
consumer.subscribe("topic2","*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt msg : list) {
//System.out.println("收到消息："+msg);
System.out.println("消息："+new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功处理, mq 收到这个 标记后相同的消息讲不会再次发给消费者
}
});
consumer.start();// 开启多线程 监控消息,持续运行
System.out.println("接收消息服务已运行");
}
}
测试：
3、单向消息
不需要有回执的消息，例如日志类消息
生产者：
public class Producer {
public static void main(String[] args) throws Exception{
DefaultMQProducer producer=new DefaultMQProducer("group1");
producer.setNamesrvAddr("192.168.23.127:9876");
producer.start();
for (int i = 1; i <= 5; i++) {
//单向消息
Message msg = new Message("topic2",("单向消息：hello rocketmq "+i).getBytes("UTF-8"));
producer.sendOneway(msg);
}
//添加一个休眠操作，确保异步消息返回后能够输出
TimeUnit.SECONDS.sleep(10);
producer.shutdown();
}
}
消费者代码同上
测试：
总结 同步消息
SendResult result = producer.send(msg);异步消息（回调处理结果必须在生产者进程结束前执行，否则回调无法正确执行）
producer.send(msg, new SendCallback() {
//表示成功返回结果
@Override
public void onSuccess(SendResult sendResult) {
System.out.println(sendResult);
}
//表示发送消息失败
@Override
public void onException(Throwable throwable) {
System.out.println(throwable);
}
});
单向消息
producer.sendOneway(msg);到此这篇关于图文并茂讲解RocketMQ消息类别的文章就介绍到这了,更多相关RocketMQ消息类别内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！
