全部产品
云市场

发送普通消息(三种方式)

更新时间:2020-07-06 19:13:18

SOFAStack 消息队列提供三种方式来发送普通消息:同步发送、异步发送和单向(Oneway)发送。本文介绍了每种发送方式的原理、使用场景、示例代码,以及三种发送方式的对比。

同步发送

原理

同步发送是指消息发送方发出一条消息后,会在收到服务端返回响应之后才发下一条消息的通讯方式。

同步发送

应用场景

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

示例代码

  1. import java.util.Properties;
  2. import com.alipay.sofa.sofamq.client.PropertyKeyConst;
  3. import io.openmessaging.api.Message;
  4. import io.openmessaging.api.MessagingAccessPoint;
  5. import io.openmessaging.api.OMS;
  6. import io.openmessaging.api.OMSBuiltinKeys;
  7. import io.openmessaging.api.Producer;
  8. import io.openmessaging.api.SendResult;
  9. public class Main {
  10. public static void main(String... args) {
  11. Properties credentials = new Properties();
  12. // AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
  13. credentials.setProperty(OMSBuiltinKeys.ACCESS_KEY, "$accessKey");
  14. // SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
  15. credentials.setProperty(OMSBuiltinKeys.SECRET_KEY, "xxxxx");
  16. // 设置 TCP 接入域名,进入控制台的概览页面查看接入点配置
  17. MessagingAccessPoint accessPoint = OMS.builder().driver("sofamq").endpoint("$endpoint")
  18. .withCredentials(credentials).build();
  19. Properties properties = new Properties();
  20. // 设置用户实例,进入控制台的概览页面查看接入点配置
  21. properties.setProperty(PropertyKeyConst.INSTANCE_ID, "$instanceId");
  22. // 您在控制台创建的 Group ID
  23. properties.setProperty(PropertyKeyConst.GROUP_ID, "YOUR_GROUP");
  24. Producer producer = accessPoint.createProducer(properties);
  25. producer.start();
  26. Message message = new Message("$topic", "YOUR_TAG", "hello world".getBytes());
  27. SendResult sendResult = producer.send(message);
  28. System.out.println(sendResult);
  29. }
  30. }

原理

异步发送是指发送方发出一条消息后,不等服务端返回响应,接着发送下一条消息的通讯方式。消息队列的异步发送,需要用户实现异步发送回调接口(SendCallback)。消息发送方在发送了一条消息后,不需要等待服务端响应即可发送第二条消息。发送方通过回调接口接收服务端响应,并处理响应结果。

异步

应用场景

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

示例代码

  1. import java.util.Properties;
  2. import com.alipay.sofa.sofamq.client.PropertyKeyConst;
  3. import io.openmessaging.api.OnExceptionContext;
  4. import io.openmessaging.api.SendCallback;
  5. import io.openmessaging.api.Message;
  6. import io.openmessaging.api.MessagingAccessPoint;
  7. import io.openmessaging.api.OMS;
  8. import io.openmessaging.api.OMSBuiltinKeys;
  9. import io.openmessaging.api.Producer;
  10. import io.openmessaging.api.SendResult;
  11. public class Main {
  12. public static void main(String... args) {
  13. Properties credentials = new Properties();
  14. // AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
  15. credentials.setProperty(OMSBuiltinKeys.ACCESS_KEY, "$accessKey");
  16. // SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
  17. credentials.setProperty(OMSBuiltinKeys.SECRET_KEY, "xxxxx");
  18. // 设置 TCP 接入域名,进入控制台的概览页面查看接入点配置
  19. MessagingAccessPoint accessPoint = OMS.builder().driver("sofamq").endpoint("$endpoint")
  20. .withCredentials(credentials).build();
  21. Properties properties = new Properties();
  22. // 设置用户实例,进入控制台的概览页面查看接入点配置
  23. properties.setProperty(PropertyKeyConst.INSTANCE_ID, "$instanceId");
  24. // 您在控制台创建的 Group ID
  25. properties.setProperty(PropertyKeyConst.GROUP_ID, "YOUR_GROUP");
  26. Producer producer = accessPoint.createProducer(properties);
  27. producer.start();
  28. Message message = new Message("$topic", "YOUR_TAG", "hello world".getBytes());
  29. // 异步发送消息, 发送结果通过 callback 返回给客户端。
  30. producer.sendAsync(message, new SendCallback() {
  31. @Override
  32. public void onSuccess(final SendResult sendResult) {
  33. // 消费发送成功
  34. System.out.println("send message success. topic=" + sendResult.getTopic() + ", msgId=" + sendResult.getMessageId());
  35. }
  36. @Override
  37. public void onException(OnExceptionContext context) {
  38. // 消息发送失败,需要进行重试处理,可重新发送这条消息或持久化这条数据进行补偿处理
  39. System.out.println("send message failed. topic=" + context.getTopic() + ", msgId=" + context.getMessageId());
  40. }
  41. });
  42. }
  43. }

单向(Oneway)发送

原理

发送方只负责发送消息,不等待服务端返回响应且没有回调函数触发,即只发送请求不等待应答。此方式发送消息的过程耗时非常短,一般在微秒级别。

oneway

应用场景

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

示例代码

  1. import java.util.Properties;
  2. import com.alipay.sofa.sofamq.client.PropertyKeyConst;
  3. import io.openmessaging.api.Message;
  4. import io.openmessaging.api.MessagingAccessPoint;
  5. import io.openmessaging.api.OMS;
  6. import io.openmessaging.api.OMSBuiltinKeys;
  7. import io.openmessaging.api.Producer;
  8. import io.openmessaging.api.SendResult;
  9. public class Main {
  10. public static void main(String... args) {
  11. Properties credentials = new Properties();
  12. // AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
  13. credentials.setProperty(OMSBuiltinKeys.ACCESS_KEY, "$accessKey");
  14. // SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
  15. credentials.setProperty(OMSBuiltinKeys.SECRET_KEY, "xxxxx");
  16. // 设置 TCP 接入域名,进入控制台的概览页面查看接入点配置
  17. MessagingAccessPoint accessPoint = OMS.builder().driver("sofamq").endpoint("$endpoint")
  18. .withCredentials(credentials).build();
  19. Properties properties = new Properties();
  20. // 设置用户实例,进入控制台的概览页面查看接入点配置
  21. properties.setProperty(PropertyKeyConst.INSTANCE_ID, "$instanceId");
  22. // 您在控制台创建的 Group ID
  23. properties.setProperty(PropertyKeyConst.GROUP_ID, "YOUR_GROUP");
  24. Producer producer = accessPoint.createProducer(properties);
  25. producer.start();
  26. Message message = new Message("$topic", "YOUR_TAG", "hello world".getBytes());
  27. producer.sendOneway(message);
  28. }
  29. }

三种发送方式的对比

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

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