开发者社区> 小旋风柴进> 正文

消息队列MQ新增3把武器

简介:
+关注继续查看

7月15日,阿里云消息中间件MQ正式发布3种消息发送方式,至此,MQ覆盖了消息发送全部场景。新增的三种方式包括可靠同步、可靠异步、单向(oneway)发送方式,满足客户在不同场景、不同传输质量和性能上的消息需求。

简介

MQ发送普通消息有三种实现方式:可靠同步发送、可靠异步发送、单向(Oneway)发送。本文介绍了每种实现的原理、使用场景以及三种实现的异同,同时提供了代码示例以供参考。

  • 可靠同步发送

    原理:同步发送是指消息发送方发出数据后,会在收到接收方发回响应之后才发下一个数据包的通讯方式。

    应用场景:此种方式应用场景非常广泛,例如重要通知邮件、报名短信通知、营销短信系统等。

  • 可靠异步发送

    原理:异步发送是指发送方发出数据后,不等接收方发回响应,接着发送下个数据包的通讯方式。MQ 的异步发送,需要用户实现异步发送回调接口(SendCallback),在执行消息的异步发送时,应用不需要等待服务器响应即可直接返回,通过回调接口接收务器响应,并对服务器的响应结果进行处理。

    应用场景:异步发送一般用于链路耗时较长,对 RT 响应时间较为敏感的业务场景,例如用户视频上传后通知启动转码服务,转码完成后通知推送转码结果等。

  • 单向(Oneway)发送

    原理:单向(Oneway)发送特点为只负责发送消息,不等待服务器回应且没有回调函数触发,即只发送请求不等待应答。此方式发送消息的过程耗时非常短,一般在微秒级别。

    应用场景:适用于某些耗时非常短,但对可靠性要求并不高的场景,例如日志收集。

下表概括了三者的特点和主要区别。

发送TPS 发送结果反馈 可靠性
同步发送 不丢失
异步发送 不丢失
单向发送 最快 可能丢失

示例代码

同步发送

public class ProducerTest {
public static void main(String[] args) {
Properties properties = new Properties();
properties.put(PropertyKeyConst.ProducerId, "XXX");//您在控制台创建的Producer ID
properties.put(PropertyKeyConst.AccessKey,"XXX");// AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.SecretKey, "XXX");// SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");//设置发送超时时间,单位毫秒
Producer producer = ONSFactory.createProducer(properties);
// 在发送消息前,必须调用start方法来启动Producer,只需调用一次即可。
producer.start();
//循环发送消息
for (int i = 0; i < 100; i++){
Message msg = new Message( //
// Message Topic
"TopicTestMQ",
// Message Tag 可理解为Gmail中的标签,对消息进行再归类,方便Consumer指定过滤条件在MQ服务器过滤
"TagA",
// Message Body 可以是任何二进制形式的数据, MQ不做任何干预,
// 需要Producer与Consumer协商好一致的序列化和反序列化方式
"Hello MQ".getBytes());
// 设置代表消息的业务关键属性,请尽可能全局唯一。
// 以方便您在无法正常收到消息情况下,可通过阿里云服务器管理控制台查询消息并补发。
// 注意:不设置也不会影响消息正常收发
msg.setKey("ORDERID_" + i);
// 同步发送消息,只要不抛异常就是成功
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
// 在应用退出前,销毁Producer对象
// 注意:如果不销毁也没有问题
producer.shutdown();
}
}

异步发送

public static void main(String[] args) {
Properties properties = new Properties();
properties.put(PropertyKeyConst.AccessKey, "DEMO_AK");// AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.SecretKey, "DEMO_SK");// SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.ProducerId, "DEMO_PID");//您在控制台创建的Producer ID
properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");//设置发送超时时间,单位毫秒
Producer producer = ONSFactory.createProducer(properties);
// 在发送消息前,必须调用start方法来启动Producer,只需调用一次即可。
producer.start();
Message msg = new Message(
// Message Topic
"TopicTestMQ",
// Message Tag,可理解为Gmail中的标签,对消息进行再归类,方便Consumer指定过滤条件在MQ服务器过滤
"TagA",
// Message Body,任何二进制形式的数据,MQ不做任何干预,需要Producer与Consumer协商好一致的序列化和反序列化方式
"Hello MQ".getBytes());
// 设置代表消息的业务关键属性,请尽可能全局唯一。以方便您在无法正常收到消息情况下,可通过MQ控制台查询消息并补发。
// 注意:不设置也不会影响消息正常收发
msg.setKey("ORDERID_100");
// 异步发送消息, 发送结果通过callback返回给客户端。
producer.sendAsync(msg, new SendCallback() {
@Override
public void onSuccess(final SendResult sendResult) {
// 消费发送成功
System.out.println("send message success. topic=" + sendResult.getTopic() + ", msgId=" + sendResult.getMessageId());
}
@Override
public void onException(OnExceptionContext context) {
// 消息发送失败
System.out.println("send message failed. topic=" + context.getTopic() + ", msgId=" + context.getMessageId());
}
});
// 在callback返回之前即可取得msgId。
System.out.println("send message async. topic=" + msg.getTopic() + ", msgId=" + msg.getMsgID());
// 在应用退出前,销毁Producer对象。注意:如果不销毁也没有问题
producer.shutdown();
}

单向(Oneway)发送

public static void main(String[] args) {
Properties properties = new Properties();
properties.put(PropertyKeyConst.AccessKey, "DEMO_AK");// AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.SecretKey, "DEMO_SK");// SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.ProducerId, "DEMO_PID");//您在控制台创建的Producer ID
properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");//设置发送超时时间,单位毫秒
Producer producer = ONSFactory.createProducer(properties);
// 在发送消息前,必须调用start方法来启动Producer,只需调用一次即可。
producer.start();
//循环发送消息
for (int i = 0; i < 100; i++){
Message msg = new Message(
// Message Topic
"TopicTestMQ",
// Message Tag,
// 可理解为Gmail中的标签,对消息进行再归类,方便Consumer指定过滤条件在MQ服务器过滤
"TagA",
// Message Body
// 任何二进制形式的数据,MQ不做任何干预,需要Producer与Consumer协商好一致的序列化和反序列化方式
"Hello MQ".getBytes());
// 设置代表消息的业务关键属性,请尽可能全局唯一。
// 以方便您在无法正常收到消息情况下,可通过阿里云服务器管理控制台查询消息并补发。
// 注意:不设置也不会影响消息正常收发
msg.setKey("ORDERID_" + i);
// oneway发送消息,只要不抛异常就是成功
producer.sendOneway(msg);
}
// 在应用退出前,销毁Producer对象
// 注意:如果不销毁也没有问题
producer.shutdown();
}

企业级互联网架构Aliware,让您的业务能力云化:https://www.aliyun.com/aliware

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
RabbitMQ:什么是消息队列MQ?为什么使用消息队列MQ?入门MQ先学哪种?(一)
MQ(Message Queue):消息队列,如今在各类业务场景中已经被广泛使用,特别在并发量日益增涨的业务和微服务架构中,消息队列能够帮助我们解决很多传统方式所不能解决的问题。 所以今天,我们就开始学习消息队列啦
117 0
同步异步调用,并谈谈消息队列mq;RocketMQ发送消息和消费消息测试类
同步调用优点: 时效性强,打电话、直播,很快可以得到结果 同步调用的问题:
116 0
4.图灵学院-----阿里/京东/滴滴/美团整理----高频MQ消息队列篇
4.图灵学院-----阿里/京东/滴滴/美团整理----高频MQ消息队列篇
51 0
秒懂消息队列MQ,万字总结带你全面了解消息队列MQ
消息队列是大型分布式系统不可缺少的中间件,也是高并发系统的基石中间件,所以掌握好消息队列MQ就变得极其重要。接下来我就将从零开始介绍什么是消息队列?消息队列的应用场景?如何进行选型?如何在Spring Boot项目中整合集成消息队列。
1211 0
盘点 常见MQ : 消息队列总览
MQ(message queue),字面上看就是一个队列,FIFO先入先出,只不过队列中存放的内容是message而已,还是一种跨进程的通信机制,用于上下游传递消息。在互联网架构中,MQ是以中国非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务。使用了MQ之后,消息发送上游只需要依赖MQ,不用依赖其他服务。
200 0
springboot整合阿里云的消息队列MQ
springboot整合阿里云的消息队列MQ
352 0
文章
问答
文章排行榜
最热
最新
相关电子书
更多
消息队列 Kafka 版差异化特性
立即下载
消息队列kafka介绍
立即下载
企业互联网架构之消息队列
立即下载