频道栏目
首页 > 程序开发 > 综合编程 > 其他综合 > 正文
RocketMQ生产者消费者示例
2017-02-07 11:22:42         来源:x763795151的博客  
收藏   我要投稿
RocketMQ生产者消费者示例:默认创建的topic读写队列数是4,用的3.5.8版本。
package com.xd.rocketmq;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;

import java.util.concurrent.TimeUnit;

/**
 * Created by dong on 2016/12/26.
 */
public class Producer {

    public static void main(String[] args) throws MQClientException, InterruptedException, RemotingException, MQBrokerException {

        final DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        //设置namesrv地址
        producer.setNamesrvAddr("192.168.100.1:9876;192.168.100.1:9876");
        producer.setInstanceName("Producer");
        // Producer对象在使用之前必须要调用start初始化,初始化一次即可
        //不可以在每次发送消息时,都调用start方法
        producer.start();
        for (int i = 0; i < 10; i++) {
            {
            Message msg = new Message("TopicTest",
                    "Tag",
                    "OrderId",
                    ("你好,TopicTest,my name is xxd_"+i).getBytes());
            System.out.println("message:"+(new String(msg.getBody())));
                System.out.println(msg);
            SendResult sendResult = producer.send(msg);
            System.out.println("Producer.main:sendResult:"+sendResult);
            }
            {
                Message msg = new Message("TopicTest1",
                        "Tag1",
                        "OrderId1",
                        ("你好,TopicTest1,my name is xxd_"+i).getBytes());
                System.out.println("message:"+(new String(msg.getBody())));
                SendResult sendResult = producer.send(msg);
                System.out.println("Producer.main:sendResult:"+sendResult);
            }
            TimeUnit.MILLISECONDS.sleep(1000);
        }
        System.out.println("group name:"+producer.getProducerGroup());
        //shutdown
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            public void run() {
                producer.shutdown();
            }
        }));
        System.exit(0);
    }
}

消费者的源代码如下:

package com.xd.rocketmq; import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer; import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently; import com.alibaba.rocketmq.client.exception.MQClientException; import com.alibaba.rocketmq.common.message.Message; import com.alibaba.rocketmq.common.message.MessageExt; import java.util.List; /** * Created by dong on 2016/12/26. */ public class Consumer { public static void main(String[] args) throws MQClientException { final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumberGroupName"); consumer.setNamesrvAddr("192.168.100.1:9876;192.168.100.1:9876"); consumer.setInstanceName("Consumer"); consumer.subscribe("TopicTest","Tag"); consumer.subscribe("TopicTest1", "*"); consumer.registerMessageListener(new MessageListenerConcurrently() { public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) { System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs); for (Message msg : msgs) { System.out.println("Consumer.consumeMessage:"+(new String(msg.getBody()))); } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); System.out.println("clientIP:"+consumer.getClientIP());; consumer.start(); System.out.println("Consumer Started."); } } 
点击复制链接 与好友分享!回本站首页
上一篇:Eclipse中将Java项目打包为jar 并且引入第三方jar包解决办法
下一篇:PowerShell的学习笔记(3)
相关文章
图文推荐
点击排行

关于我们 | 联系我们 | 广告服务 | 投资合作 | 版权申明 | 在线帮助 | 网站地图 | 作品发布 | Vip技术培训 | 举报中心

版权所有: 红黑联盟--致力于做实用的IT技术学习网站