文档首页/ 分布式消息服务RocketMQ版/ 开发指南/ Java（TCP协议）/ 收发普通消息
更新时间：2024-12-10 GMT+08:00
查看PDF
分享

收发普通消息

本章节介绍普通消息的收发方法和示例代码。其中，普通消息发送方式分为同步发送和异步发送。

  • 同步发送：消息发送方发出一条消息到服务端，服务端接收并处理消息，然后返回响应给发送方，发送方收到响应后才会发送下一条消息。
  • 异步发送：消息发送方发出一条消息后，不等服务端返回响应，接着发送下一条消息。

收发普通消息前，请参考收集连接信息收集RocketMQ所需的连接信息。

约束与限制

客户端连接RocketMQ实例5.x版本收发普通消息前，需要确保Topic的消息类型为“普通”。

准备环境

开源的Java客户端支持连接分布式消息服务RocketMQ版，推荐使用的客户端版本为4.9.8。

通过以下任意一种方式引入依赖：
  • 使用Maven方式引入依赖。 
    <dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.9.8</version>
</dependency>
  • 下载依赖JAR包。

同步发送

同步发送是指消息发送方发出一条消息到服务端，服务端接收并处理消息，然后返回响应给发送方，发送方收到响应后才会发送下一条消息的通讯方式。

参考如下示例代码，或者通过Producer.java获取更多示例代码。

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class Main {
    public static void main(String[] args) {
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        //填入连接地址
        producer.setNamesrvAddr("192.168.0.1:8100");
        //producer.setUseTLS(true); //创建实例时，如果开启了SSL，请添加此行代码。
        try {
            producer.start();
            Message msg = new Message("TopicTest",
                "TagA",
                "OrderID188",
                "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);

        } catch (Exception e) {
            e.printStackTrace();

        }
        producer.shutdown();
    }
}

异步发送

异步发送是指消息发送方发出一条消息后，不等服务端返回响应，接着发送下一条消息的通讯方式。

使用异步发送需要客户端实现异步发送回调接口（SendCallback）。即消息发送方在发送了一条消息后，不需要等待服务端响应接着发送第二条消息。发送方通过回调接口接收服务端响应，并处理响应结果。

参考如下示例代码，或者通过AsyncProducer.java获取更多示例代码。

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;


public class Main {
    public static void main(String[] args) throws InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        //填入连接地址
        producer.setNamesrvAddr("192.168.120.45:8100;192.168.123.150:8100");
        //producer.setUseTLS(true);    //创建实例时，如果开启了SSL，请增加此行代码。
        try {
            producer.start();
            Message msg = new Message("TopicTest",
                    "TagA",
                    "OrderID188",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult result) {
                    // 消息发送成功。
                    System.out.println("send message success. msgId= " + result.getMsgId());
                }


                @Override
                public void onException(Throwable throwable) {
                    // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理。
                    System.out.println("send message failed.");
                    throwable.printStackTrace();
                }
            });


        } catch (Exception e) {
            e.printStackTrace();
        }
        Thread.sleep(2000);
        producer.shutdown();
    }}

订阅普通消息

参考如下示例代码，或者通过PushConsumer.java获取更多示例代码。

import java.util.List;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

public class PushConsumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
        //填入连接地址
        consumer.setNamesrvAddr("192.168.0.1:8100");
        //consumer.setUseTLS(true);    //创建实例时，如果开启了SSL，请添加此行代码。
        consumer.subscribe("TopicTest", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

父主题： Java（TCP协议）

相关文档