阿里云首页 金融分布式架构 SOFAStack

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

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

同步发送

原理

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

同步发送

应用场景

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

示例代码

import java.util.Properties;
import com.alipay.sofa.sofamq.client.PropertyKeyConst;
import io.openmessaging.api.Message;
import io.openmessaging.api.MessagingAccessPoint;
import io.openmessaging.api.OMS;
import io.openmessaging.api.OMSBuiltinKeys;
import io.openmessaging.api.Producer;
import io.openmessaging.api.SendResult;

public class Main{
    public static void main(String... args){
        Properties credentials =new Properties();
        // AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
        credentials.setProperty(OMSBuiltinKeys.ACCESS_KEY,"$accessKey");
        // SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
        credentials.setProperty(OMSBuiltinKeys.SECRET_KEY,"xxxxx");
        
        // 设置 TCP 接入域名,进入控制台的概览页面查看接入点配置
        MessagingAccessPoint accessPoint = OMS.builder().driver("sofamq").endpoint("$endpoint")
        .withCredentials(credentials).build();
        
        Properties properties =new Properties();
        // 设置用户实例,进入控制台的概览页面查看接入点配置
        properties.setProperty(PropertyKeyConst.INSTANCE_ID,"$instanceId");
        // 您在控制台创建的 Group ID
        properties.setProperty(PropertyKeyConst.GROUP_ID,"YOUR_GROUP");
        Producer producer = accessPoint.createProducer(properties);
        
        producer.start();
        
        Message message =new Message("$topic","YOUR_TAG","hello world".getBytes());
        SendResult sendResult = producer.send(message);
        System.out.println(sendResult);
    }
}

原理

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

异步

应用场景

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

示例代码

import java.util.Properties;
import com.alipay.sofa.sofamq.client.PropertyKeyConst;
import io.openmessaging.api.OnExceptionContext;
import io.openmessaging.api.SendCallback;
import io.openmessaging.api.Message;
import io.openmessaging.api.MessagingAccessPoint;
import io.openmessaging.api.OMS;
import io.openmessaging.api.OMSBuiltinKeys;
import io.openmessaging.api.Producer;
import io.openmessaging.api.SendResult;

public class Main {
    public static void main(String... args) {
        Properties credentials = newProperties();
        // AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
        credentials.setProperty(OMSBuiltinKeys.ACCESS_KEY, "$accessKey");
        // SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
        credentials.setProperty(OMSBuiltinKeys.SECRET_KEY, "xxxxx");

        // 设置 TCP 接入域名,进入控制台的概览页面查看接入点配置
        MessagingAccessPoint accessPoint = OMS.builder().driver("sofamq").endpoint("$endpoint")
                .withCredentials(credentials).build();

        Properties properties = newProperties();
        // 设置用户实例,进入控制台的概览页面查看接入点配置
        properties.setProperty(PropertyKeyConst.INSTANCE_ID, "$instanceId");
        // 您在控制台创建的 Group ID
        properties.setProperty(PropertyKeyConst.GROUP_ID, "YOUR_GROUP");
        Producer producer = accessPoint.createProducer(properties);

        producer.start();

        Message message = new Message("$topic", "YOUR_TAG", "hello world".getBytes());
        // 异步发送消息, 发送结果通过 callback 返回给客户端。
        producer.sendAsync(message, newSendCallback() {
            @Override
            public void onSuccess (finalSendResult 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());
            }
        });
    }
}

单向(Oneway)发送

原理

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

oneway

应用场景

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

示例代码

import java.util.Properties;
import com.alipay.sofa.sofamq.client.PropertyKeyConst;
import io.openmessaging.api.Message;
import io.openmessaging.api.MessagingAccessPoint;
import io.openmessaging.api.OMS;
import io.openmessaging.api.OMSBuiltinKeys;
import io.openmessaging.api.Producer;
import io.openmessaging.api.SendResult;

public class Main {
    public static void main(String... args) {
        Properties credentials = new Properties();
        // AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
        credentials.setProperty(OMSBuiltinKeys.ACCESS_KEY, "$accessKey");
        // SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
        credentials.setProperty(OMSBuiltinKeys.SECRET_KEY, "xxxxx");

        // 设置 TCP 接入域名,进入控制台的概览页面查看接入点配置
        MessagingAccessPoint accessPoint = OMS.builder().driver("sofamq").endpoint("$endpoint")
                .withCredentials(credentials).build();

        Properties properties = new Properties();
        // 设置用户实例,进入控制台的概览页面查看接入点配置
        properties.setProperty(PropertyKeyConst.INSTANCE_ID, "$instanceId");
        // 您在控制台创建的 Group ID
        properties.setProperty(PropertyKeyConst.GROUP_ID, "YOUR_GROUP");
        Producer producer = accessPoint.createProducer(properties);

        producer.start();

        Message message = new Message("$topic", "YOUR_TAG", "hello world".getBytes());
        producer.sendOneway(message);
    }
}

三种发送方式的对比

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

发送方式

发送 TPS

发送结果反馈

可靠性

同步发送

不丢失

异步发送

不丢失

单向发送

最快

可能丢失