RocketMQ 4.1.0 与 Docker搭建与集群
2017-05-25 17:13
615 查看
搭建环境:docker
Centos7 RocketMQ 4.1.0 java 1.8
这里提供 编译好的 (RocketMQ4.1) apache-rocketmq-all.tar.gz 下载
ps:如果你没
apache-rocketmq-all.tar.gz 请在 搭建环境里 (补充 git 和 maven )
git clone https://github.com/apache/incubator-rocketmq.git mvn clean package install -Prelease-all assembly:assembly -U
--案例只适合测试学习
docker配置:
容器1:9876/tcp localhost:9876
10909/tcp localhost:10909
10911/tcp localhost:10911
容器2:
9876/tcp localhost:9879
10909/tcp localhost:10916
10915/tcp localhost:10915
--查询宿主机的ip
例如是 os 系统:ifconfig -a 找到 en0 -》net 192.168.1.7
ps:linux 也是一样
我在docker 挂载的盘在 /sys/fs/cgroup/
1.上传解压【两台机器】
# cd ./usr # mkdir rockerMQ # cd ./rockerMQ # tar -zxvf /sys/fs/cgroup/apache-rocketmq-all.tar.gz --这里建一个软连接(在最外面) #ln -s /usr/rockerMQ rockermq
2.创建存储路径【两台机器】(可以用默认)
# mkdir /sys/fs/cgroup/store # mkdir /sys/fs/cgroup/store/commitlog # mkdir /sys/fs/cgroup/store/consumequeue # mkdir /sys/fs/cgroup/store/index # mkdir /sys/fs/cgroup/store/logs
3.修改日志配置文件【两台机器】
# cd /usr/rocketMQ/conf # sed -i 's#${user.home}#/sys/fs/cgroup/store#g' *.xml --ps:为了减少冲突 #cd /usr/rocketMQ/bin #sed -i 's/\r//g' *ps: sed -i 's#${user.home}#/sys/fs/cgroup/store#g' *.xml
4.修改启动脚本参数【两台机器】
#cd /usr/rocketMQ/bin #vi runbroker.sh ----修改 JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512m - XX:PermSize=128m -XX:MaxPermSize=320m #vi runserver.sh ----修改JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512m - XX:PermSize=128m -XX:MaxPermSize=320m
5.RocketMQ配置文件【两台机器】
# cd /usr/rocketMQ/conf/2m-noslave --修改 容器1 #vi broker-a.properties --修改 容器2 #vi broker-b.properties --建议两个容器都改
配置详情:
#所属集群名字 brokerClusterName=DefaultCluster #broker名字,注意此处不同的配置文件填写的不一样 brokerName=broker-a|broker-b #0 表示 Master,>0 表示 Slave brokerId=0 #nameServer地址,分号分割 namesrvAddr=192.168.1.107:9876;192.168.1.107:9879 #关键对于docker brokerIP1=192.168.1.107 brokerIP2=192.168.1.107 #在发送消息时,自动创建服务器不存在的topic,默认创建的队列数 defaultTopicQueueNums=4 #是否允许 Broker 自动创建Topic,建议线下开启,线上关闭 autoCreateTopicEnable=true #是否允许 Broker 自动创建订阅组,建议线下开启,线上关闭 autoCreateSubscriptionGroup=true #Broker 对外服务的监听端口 listenPort=10911 #删除文件时间点,默认凌晨 4点 deleteWhen=04 #文件保留时间,默认 48 小时 fileReservedTime=120 #commitLog每个文件的大小默认1G mapedFileSizeCommitLog=1073741824 #ConsumeQueue每个文件默认存30W条,根据业务情况调整 mapedFileSizeConsumeQueue=300000 #destroyMapedFileIntervalForcibly=120000 #redeleteHangedFileInterval=120000 #检测物理文件磁盘空间 diskMaxUsedSpaceRatio=88 #这里是我的 日志配置 #存储路径 storePathRootDir=/sys/fs/cgroup/store #commitLog 存储路径 storePathCommitLog=/usr/local/rocketmq/store/commitlog #消费队列存储路径存储路径 storePathConsumeQueue=/sys/fs/cgroup/store/consumequeue #消息索引存储路径 storePathIndex=/sys/fs/cgroup/store/index #checkpoint 文件存储路径 storeCheckpoint=/sys/fs/cgroup/store/checkpoint #abort 文件存储路径 abortFile=/sys/fs/cgroup/store/abort #限制的消息大小 maxMessageSize=65536 #flushCommitLogLeastPages=4 #flushConsumeQueueLeastPages=2 #flushCommitLogThoroughInterval=10000 #flushConsumeQueueThoroughInterval=60000 #Broker 的角色 #- ASYNC_MASTER 异步复制Master #- SYNC_MASTER 同步双写Master #- SLAVE brokerRole=ASYNC_MASTER #刷盘方式 #- ASYNC_FLUSH 异步刷盘 #- SYNC_FLUSH 同步刷盘 flushDiskType=ASYNC_FLUSH #checkTransactionMessageEnable=false #发消息线程池数量 #sendMessageThreadPoolNums=128 #拉消息线程池数量 #pullMessageThreadPoolNums=128
ps:注意上面 broker
如:容器1:brokerName=broker-a listenPort=10911
容器2:brokerName=broker-b listenPort=10915
准备工作完成了~~~
6.启动NameServer【两台机器】
#cd /usr/rocketMQ/bin #nohup sh mqnamesrv & #jps #tail -f -n 500 /sys/fs/cgroup/store/logs/rocketmqlogs/namesrv.log
7.启动BrokerServer
#nohup sh mqbroker -c /rocketmq/conf/2m-noslave/broker-a.properties >/dev/null 2>&1 & --另外一个容器 #nohup sh mqbroker -c /rocketmq/conf/2m-noslave/broker-b.properties >/dev/null 2>&1 & #jps #tail -f -n 500 /sys/fs/cgroup/store/logs/rocketmqlogs/broker.log
8.关闭
#cd /usr/rocketMQ/bin #sh mqshutdown broker #sh mqshutdown namers #jps
java测试代码:
--product
import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import java.util.concurrent.TimeUnit; public class Producer { public static void main(String[] args) throws MQClientException, InterruptedException { /** * 一个应用创建一个Producer,由应用来维护此对象,可以设置为全局对象或者单例<br> * 注意:ProducerGroupName需要由应用来保证唯一<br> * ProducerGroup这个概念发送普通的消息时,作用不大,但是发送分布式事务消息时,比较关键, * 因为服务器会回查这个Group下的任意一个Producer */ DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName"); producer.setNamesrvAddr("192.168.1.7:9876;192.168.1.7:9879"); producer.setInstanceName("Producer"); producer.setVipChannelEnabled(false); /** * Producer对象在使用之前必须要调用start初始化,初始化一次即可<br> * 注意:切记不可以在每次发送消息时,都调用start方法 */ producer.start(); /** * 下面这段代码表明一个Producer对象可以发送多个topic,多个tag的消息。 * 注意:send方法是同步调用,只要不抛异常就标识成功。但是发送成功也可会有多种状态,<br> * 例如消息写入Master成功,但是Slave不成功,这种情况消息属于成功,但是对于个别应用如果对消息可靠性要求极高,<br> * 需要对这种情况做处理。另外,消息可能会存在发送失败的情况,失败重试由应用来处理。 */ for (int i = 0; i < 2; i++) { try { { Message msg = new Message("TopicTest1",// topic "TagA",// tag "OrderID001",// key ("Hello MetaQ-A"+i).getBytes());// body SendResult sendResult = producer.send(msg); System.out.println(sendResult); } { Message msg = new Message("TopicTest2",// topic "TagB",// tag "OrderID0034",// key ("Hello MetaQ-B"+i).getBytes());// body SendResult sendResult = producer.send(msg); System.out.println(sendResult); } { Message msg = new Message("TopicTest3",// topic "TagC",// tag "OrderID061",// key ("Hello MetaQ-C"+i).getBytes());// body SendResult sendResult = producer.send(msg); System.out.println(sendResult); } } catch (Exception e) { e.printStackTrace(); } TimeUnit.MILLISECONDS.sleep(1000); } /** * 应用退出时,要调用shutdown来清理资源,关闭网络连接,从MetaQ服务器上注销自己 * 注意:我们建议应用在JBOSS、Tomcat等容器的退出钩子里调用shutdown方法 */ producer.shutdown(); } }
--Consumer
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.message.MessageExt; import java.util.List; 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.7:9876;192.168.1.7:9879"); 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() { /** * 默认msgs里只有一条消息,可以通过设置consumeMessageBatchMaxSize参数来批量接收消息 */ @Override 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 && msg.getTags().equals("TagA")) { // 执行TagA的消费 System.out.println(new String(msg.getBody())); } else if (msg.getTags() != null && msg.getTags().equals("TagC")) { // 执行TagC的消费 } else if (msg.getTags() != null && msg.getTags().equals("TagD")) { // 执行TagD的消费 } } else if (msg.getTopic().equals("TopicTest2")) { System.out.println(new String(msg.getBody())); } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); /** * Consumer对象在使用之前必须要调用start初始化,初始化一次即可<br> */ consumer.start(); System.out.println("Consumer Started."); } }
相关文章推荐
- docker搭建nginx+springboot集群
- 使用Docker在本地搭建Hadoop分布式集群
- Docker学习笔记 — Swarm搭建Docker集群
- Docker学习笔记 — Swarm搭建Docker集群
- Docker1.12.1之swarm集群搭建与使用
- centos7搭建Docker Swarm 集群
- Docker + Swarm + etcd 集群搭建
- Docker Kubernetes1.5.2集群环境搭建(CentOS)
- 基于Docker快速搭建多节点Hadoop集群--已验证
- 使用docker搭建zookeeper集群
- 用docker搭建elasticsearch集群
- Swarm搭建Docker集群
- docker应用-6(mysql+mycat 搭建数据库集群)
- docker搭建zookeeper集群
- 【Docker环境搭建Zookeeper集群】
- 利用Docker搭建MongoDB副本集群
- 使用 Docker搭建 ZooKeeper 集群
- Docker教程之 Swarm搭建Docker集群
- Docker下的Kafka学习之二:搭建集群环境
- 基于Windows/Linux混合搭建Docker集群架构