一、启动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 node1
192.168.1.103 master
192.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:9876
BenchmarkTest
DefaultCluster
SELF_TEST_TOPIC
%RETRY%please_rename_unique_group_name_4
TBW102
gaojingsong
master
OFFSET_MOVED_EVENT
[root@master bin]# cd ../
备注:此时topic不存在,但是生产数据的时候会自动创建
二、生产和消费数据
生产:
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);
}
}
消费:
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.");
}
}
三、验证消费结果
[root@master bin]# sh mqadmin topicList -n 192.168.1.103:9876
BenchmarkTest
TopicTest1
DefaultCluster
SELF_TEST_TOPIC
%RETRY%please_rename_unique_group_name_4
%RETRY%ConsumerGroupName
TBW102
gaojingsong
master
OFFSET_MOVED_EVENT
[root@master bin]# sh mqadmin topicStatus -n 192.168.1.103:9876 -t TopicTest1
#Broker Name #QID #Min Offset #Max Offset #Last Updated
master 0 0 4 2016-10-20 14:38:19,236
master 1 0 4 2016-10-20 14:38:19,243
master 2 0 2 2016-10-20 14:38:15,171
master 3 0 2 2016-10-20 14:38:15,180
[root@master bin]# shutdown -h now
消费数据
四、错误解决方案
解决方案:版本不一致问题,更改同一版本问题解决
相关推荐
RocketMQ生产者和消费者Java代码示例
JAVA架构RocketMQ单机环境搭建pdf+架构师视频资源
java多线程实现生产者和消费者
java操作rocketmq4.7,jar包和linux里安装的rocketmq版本要一致;有独立的生产者和消费者
这是我辛辛苦苦写的java代码,关于生产消费的列子,非常好的
基于java springboot的rocketmq生产和消费demo,开发工具为idea
JAVA-ACE-架构师系列视频教程RocketMQ订单实战上目录介绍: 1001_RocketMQ_简介 2002_RocketMQ_核心概念详解 3003_RocketMQ_集群构建模型详解(一) 4004_RocketMQ_集群构建模型详解(二) 5005_RocketMQ_双主模式...
java生产者消费者问题 java生产者消费者问题 java生产者消费者问题 java生产者消费者问题 java生产者消费者问题
rocketMQ测试程序java版本。直接运行main方法。
java的生产者消费者,thread生产者消费者、。。。。。。
rocketmq控制台jar和启动指令,用来图形化查看mq队列和消息,启动指令:nohup java -jar rocketmq-console-ng-1.0.0.jar --server.port=12581 --rocketmq.config.namesrvAddr=192.168.100.62:28881 & 或者 nohup ...
java 使用 rocketmq的一个生产者和消费者的实现,其中要先启动rocket的nameserver 和borker
完整的jar,建立Java项目,1.8环境, 添加到路径则可以运行。包含非mqjar。common,netty的。
Rocketmq-AvroDemo按照avro规范向rocketmq中生产和消费数据按照avro格式生成数据并且写入rocketmq,再根据schema在rocketmq中消费avro格式的数据。
即时显示有界缓冲区的全部内容,当前指针位置和生产者/消费者县城的标识符.(2)生产者和消费者各有两个以上.(3)多个生产者或多个消费者之间须有共享对缓冲区进行操作的函数代码. 提示:(1) 有界缓冲区可用数组实现.
java实现生产并消费rocketmq数据并将数据上传到存储服务器上 源码。 下载该项目可以学到 rocketmq 生产和消费数据。 上传数据到阿里云oss 和本地minio上 minio版本用的是8.3.0
java实现生产并消费rocketmq数据并将数据上传到存储服务器上 源码。 下载该项目可以学到 rocketmq 生产和消费数据。 上传数据到阿里云oss 和本地minio上 minio版本用的是8.3.0
操作系统生产者与消费者java源代码,使用Java图形用户界面实现。
用java编写,模拟生产者消费者问题,生产一个消费一个!