RocketMQ中哪位大佬有C/C++调java版consumer的例子？-[阿里云_云淘科技]

RocketMQ中哪位大佬有C/C++调java版consumer的例子？

以下为热心网友提供的参考意见

Apache RocketMQ的消费者可以使用Java客户端来实现。以下是一个简单的示例代码，展示了如何使用DefaultMQPushConsumer来消费消息：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
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.common.message.MessageExt;

public class Consumer {
    public static void main(String[] args) throws MQClientException {
        // 创建一个名为 "ConsumerGroupName" 的消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        // 订阅一个名为 "TopicTest" 的主题
        consumer.subscribe("TopicTest", "*");
        // 注册一个消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("Receive message: " + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者
        consumer.start();
        System.out.println("Consumer started.");
    }
}

以下为热心网友提供的参考意见

直接java客户端消费 通过http协议发给C呗，c有一个客户端的， 基本功能都没有问题https://github.com/apache/rocketmq-client-cpp ，此回答整理自钉群“群2-Apache RocketMQ 中国开发者钉钉群”

以下为热心网友提供的参考意见

对于C/C++调用Java版的RocketMQ Consumer，Apache RocketMQ官方并未提供直接的示例。但你可以借鉴其Java客户端的使用方式，使用JNI的方式进行调用。

在RocketMQ中，消息消费者（Consumer）负责消费消息，一般是后台系统负责异步消费。一个消息消费者会从Broker服务器拉取消息、并将其提供给应用程序。以下是一个简单的Java客户端消费消息的示例代码：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
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.common.message.MessageExt;

public class Consumer {
    public static void main(String[] args) throws MQClientException {
        // 创建一个名为 "ConsumerGroupName" 的消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");

        // 设置 NameServer 地址
        consumer.setNamesrvAddr("127.0.0.1:9876");

        // 订阅一个名为 "TopicTest" 的主题
        consumer.subscribe("TopicTest", "*");

        // 注册一个消息监听器，用于处理从 Broker 接收到的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msg.getBody()));
                }
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者实例
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}