JAVA代码之RocketMQ生产和消费数据
2016-10-22 11:18
302 查看
一、启动RocketMQ[root@master ~]# cat /etc/hosts# Do not remove the following line, or various programs# that require network functionality will fail.127.0.0.1 localhost.localdomain localhost::1 localhost6.localdomain6 localhost6 192.168.1.106 node1192.168.1.103 master192.168.1.110 node2[root@master ~]# cd /opt/alibaba-rocketmq/bin/[root@master bin]# cat play.sh #!/bin/sh## Name Server#nohup sh mqnamesrv > ns.log 2>&1 # Service Addr#ADDR=`hostname -i`:9876## Broker#nohup sh mqbroker -n ${ADDR} > bk.log 2>&1 & echo "Start Name Server and Broker Successfully, ${ADDR}"[root@master bin]# sh play.sh Start Name Server and Broker Successfully, 192.168.1.103:9876[root@master bin]# sh mqadmin topicList -n 192.168.1.103:9876BenchmarkTestDefaultClusterSELF_TEST_TOPIC%RETRY%please_rename_unique_group_name_4TBW102gaojingsongmasterOFFSET_MOVED_EVENT[root@master bin]# cd ../备注:此时topic不存在,但是生产数据的时候会自动创建 二、生产和消费数据生产:[b]下载[/b]package cn.cn.mq.demo; import java.util.concurrent.TimeUnit; 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; public class Producer { public static void main(String[] args) throws MQClientException, InterruptedException{ /** * 一个应用创建一个Producer,由应用来维护此对象,可以设置为全局对象或者单例<br> * 注意:ProducerGroupName需要由应用来保证唯一<br> * ProducerGroup这个概念发送普通的消息时,作用不大,但是发送分布式事务消息时,比较关键, * 因为服务器会回查这个Group下的任意一个Producer */ final DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName"); producer.setNamesrvAddr("192.168.1.103:9876"); producer.setInstanceName("Producer"); /** * Producer对象在使用之前必须要调用start初始化,初始化一次即可<br> * 注意:切记不可以在每次发送消息时,都调用start方法 */ producer.start(); /** * 下面这段代码表明一个Producer对象可以发送多个topic,多个tag的消息。 * 注意:send方法是同步调用,只要不抛异常就标识成功。但是发送成功也可会有多种状态,<br> * 例如消息写入Master成功,但是Slave不成功,这种情况消息属于成功,但是对于个别应用如果对消息可靠性要求极高,<br> * 需要对这种情况做处理。另外,消息可能会存在发送失败的情况,失败重试由应用来处理。 */ for (int i = 0; i < 3; i++){ try { { Message msg = new Message("TopicTest1",// topic "TagA",// tag "OrderID001",// key ("我的名字是程序员:"+i).getBytes());// body SendResult sendResult = producer.send(msg); System.out.println(sendResult); } { Message msg = new Message("TopicTest1",// topic "TagC",// tag "OrderID001",// key ("我来测试RocketMQ:"+i).getBytes());// body SendResult sendResult = producer.send(msg); System.out.println(sendResult); } }catch(Exception e) { e.printStackTrace(); } TimeUnit.MILLISECONDS.sleep(4000); } /** * 应用退出时,要调用shutdown来清理资源,关闭网络连接,从MetaQ服务器上注销自己 * 注意:我们建议应用在JBOSS、Tomcat等容器的退出钩子里调用shutdown方法 */// producer.shutdown(); Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() { public void run() { producer.shutdown(); } })); System.exit(0); }} 消费:[b]下载[/b]package cn.cn.mq.demo; import java.util.List; 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.MessageExt; public class PushConsumer {/*** 当前例子是PushConsumer用法,使用方式给用户感觉是消息从RocketMQ服务器推到了应用客户端。<br>* 但是实际PushConsumer内部是使用长轮询Pull方式从MetaQ服务器拉消息,然后再回调用户Listener方法<br>*/public static void main(String[] args) throws InterruptedException,MQClientException {/*** 一个应用创建一个Consumer,由应用来维护此对象,可以设置为全局对象或者单例<br>* 注意:ConsumerGroupName需要由应用来保证唯一*/DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");consumer.setNamesrvAddr("192.168.1.103:9876");consumer.setInstanceName("Consumber"); /*** 订阅指定topic下tags分别等于TagA或TagC或TagD*/consumer.subscribe("TopicTest1", "TagA || TagC || TagD");/*** 订阅指定topic下所有消息<br>* 注意:一个consumer对象可以订阅多个topic*///consumer.subscribe("TopicTest2", "*"); consumer.registerMessageListener(new MessageListenerConcurrently() { public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { System.out.println(Thread.currentThread().getName()+ " Receive New Messages: " + msgs.size()); MessageExt msg = msgs.get(0);if (msg.getTopic().equals("TopicTest1")) {// 执行TopicTest1的消费逻辑if (msg.getTags() != null // 执行TagA的消费&& msg.getTags().equals("TagA")) {System.out.println("TagA:"+new String(msg.getBody()));} else if (msg.getTags() != null// 执行TagC的消费&& msg.getTags().equals("TagC")) {System.out.println("TagC:"+new String(msg.getBody()));} else if (msg.getTags() != null// 执行TagD的消费&& msg.getTags().equals("TagD")) {System.out.println("TagD:"+new String(msg.getBody()));}} return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; }}); /*** Consumer对象在使用之前必须要调用start初始化,初始化一次即可<br>*/consumer.start(); System.out.println("ConsumerStarted.");}} 三、验证消费结果 [b]下载[/b][root@master bin]# sh mqadmin topicList -n 192.168.1.103:9876BenchmarkTestTopicTest1DefaultClusterSELF_TEST_TOPIC%RETRY%please_rename_unique_group_name_4%RETRY%ConsumerGroupNameTBW102gaojingsongmasterOFFSET_MOVED_EVENT[root@master bin]# sh mqadmin topicStatus -n 192.168.1.103:9876 -t TopicTest1#Broker Name #QID #Min Offset #Max Offset #Last Updatedmaster 0 0 4 2016-10-20 14:38:19,236master 1 0 4 2016-10-20 14:38:19,243master 2 0 2 2016-10-20 14:38:15,171master 3 0 2 2016-10-20 14:38:15,180[root@master bin]# shutdown -h now
![](http://dl2.iteye.com/upload/attachment/0120/7196/4e43d721-8420-3b09-820d-5b2b0ffb5242.png)
消费数据
![](http://dl2.iteye.com/upload/attachment/0120/7198/ba28aa80-f91d-3e08-9dac-b245e9335c00.png)
四、错误解决方案 [b]下载[/b]
![](http://dl2.iteye.com/upload/attachment/0120/7196/4e43d721-8420-3b09-820d-5b2b0ffb5242.png)
消费数据
![](http://dl2.iteye.com/upload/attachment/0120/7198/ba28aa80-f91d-3e08-9dac-b245e9335c00.png)
四、错误解决方案 [b]下载[/b]
![](http://dl2.iteye.com/upload/attachment/0120/7200/0e4394dd-3452-3c7c-a59e-d9e1ce7b2751.png)
相关文章推荐
- JAVA代码之RocketMQ生产和消费数据
- 《黑马程序员》java笔记->生产消费多线程,同步代码块(优化代码)
- Kafka 使用Java实现数据的生产和消费demo
- 简单生产消费模式的代码流程(Java代码)
- JAVA生产消费模式(代码及流程图)
- JAVA 根据数据库表内容生产树结构JSON数据的实例代码
- Kafka利用Java实现数据的生产和消费实例教程
- Kafka 使用Java实现数据的生产和消费demo
- Java连接MQ报错,通道协商失败 MQJE001: 完成代码为“2”,原因为“2059
- 从XML到Java代码的数据绑定之二
- 在Struts中使用JavaBean和List(多行数据)类型属性-JSP教程,Java技巧及代码
- 实例代码摘录_生产消费、死锁
- java mapping生产静态页面核心代码
- JAVA功能代码《9》----Java发送HTTP请求和提取数据
- [Web-Harvest数据采集之二]Web-Harvest基础-抓取java代码分析
- JAVA读取txt文件中数据代码范例
- 分享非常有用的Java程序(关键代码)(八)---Java InputStream读取网络响应Response数据的方法!(重要)
- java代码用于返回JSON或者XML数据(extJs)
- java生产消费线程小例子
- java数据类型转化大全实例代码