Kafka,RocketMQ,RabbitMQ部署与使用体验
前言
近期在研究各种消息队列方案,为了有一个直观的使用体验,我把Kafka,RocketMQ,RabbitMQ各自部署了一遍,并使用了最基本的生产与消费消息功能。在部署过程中也遇到一些问题,特此记录。本文只适用于没有使用过消息队列,还停留在安装部署阶段的新手用户,要了解一个软件,最好的开始方法是开始使用他,这样才会有一个直观的印象。本篇文章的作用也在于此,至于需要了解更深入的架构与细节,则需要查询其他的文档资料,这也不是本文的目的。我这里使用的操作系统是Centos 6.x,硬件配置一般即可。
Kafka的部署与使用
Kafka的部署我是参考官网的步骤开始的,请直接参考其Quickstart章节。
Step1:下载安装包并解压
wget https://mirror.bit.edu.cn/apache/kafka/2.4.1/kafka_2.12-2.4.1.tgz tar -xzf kafka_2.12-2.4.1.tgz cd kafka_2.12-2.4.1
问题:下载有可能报下面的错误(没有报错则忽略):
To connect to mirrors.tuna.tsinghua.edu.cn insecurely, use ‘–no-check-certificate’.
只需要添加报错提示的参数即可:
wget --no-check-certificate https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.4.1/kafka_2.12-2.4.1.tgz
Step2:启动服务器
在上述第一步执行完成之后,当前在目录kafka_2.12-2.4.1下,运行下面的命令启动zookeeper服务,Kafka依赖zookeeper服务,所以我们首先要启动zookeeper服务,由于zookeeper服务需要另外安装,Kafka的安装包中提供了一个简单的单节点的zookeeper实例,以方便快速运行Kafka服务,如果是在正式环境中使用则需要另外单独安装zookeeper服务(一般是分布式集群而非单节点)给Kafka使用。
bin/zookeeper-server-start.sh config/zookeeper.properties
我在运行此命令时候报下面的错误:
Unrecognized VM option \’+UseGCLogFileRotation\’
Could not create the Java virtual machine.
从提示上看是因为java虚拟机不支持参数UseGCLogFileRotation,我经过查资料说这个参数可以去掉,bin
/zookeeper-server-start
.sh 脚本中调用了 bin/kafka-run-class.sh ,真正起作用的代码在这个脚本中 kafka-run-class.sh,打开这个脚本找到上述参数所在的地方,去掉 -XX:+UseGCLogFileRotation , 如下图所示:
继续运行上面的脚本,继续报错:
Unrecognized VM option \’NumberOfGCLogFiles=10\’
Could not create the Java virtual machine.
像刚才一样,继续去掉 -XX:NumberOfGCLogFiles=10 ,如下图:
继续运行,得到报错:
Unrecognized VM option \’GCLogFileSize=100M\’
Could not create the Java virtual machine.
我们继续去掉脚本中的 -XX:GCLogFileSize=100M 如下图:
继续运行,得到下面的报错:
从提示信息可以看出应该是java字节码的兼容性问题,java.lang.UnsupportedClassVersionError,表示当前运行的zookeeper的字节码是高版本的java编译器编译的,而现在的运行时环境是低版本,所以不兼容。运行下面的命令查看java版本:
java -version java version "1.6.0_24" OpenJDK Runtime Environment (IcedTea6 1.11.8) (rhel-1.56.1.11.8.el6_3-x86_64) OpenJDK 64-Bit Server VM (build 20.0-b12, mixed mode)
是jdk1.6,这个版本太低了,我们需要安装jdk1.8或者以上的版本,怎么安装呢,这里我首先搜索一下系统的安装源里面有没有,如果有我们直接从操作系统提供的源进行安装,如果没有则需要通过源码等其他办法进行安装了,运行下面命令查看:
yum search openjdk
结果图如下:
我们安装开发环境,也就是下面这个,区别就是开发环境中有javac等编译工具,而运行时环境中没有,运行时环境只包含运行java程序所需要的工具,因为后面我们需要使用javac编译,所以我们安装开发环境,使用下面命令安装完成即可:
yum install java-1.8.0-openjdk-devel.x86_64
成功之后我们再次运行 bin
/zookeeper-server-start
.sh config
/zookeeper
.properties 以启动zookeeper服务,没有任何错误,最后停留在如下界面:
启动Kafka服务:bin
/kafka-server-start
.sh config
/server
.properties 没有任何报错,中间会打印出Kafka的配置,输出比较多,我只截取了一部分,其中 INFO KafkaConfig values 后面就是kafka的配置:
Step3:创建topic
bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic testtopic
我们可以列出kafka服务器上的topic列表:
bin/kafka-topics.sh --list --bootstrap-server localhost:9092 testtopic
Step4:发送消息
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testtopic
>hello message
>
运行命令之后 > 提示符等待输入消息,我们输入字符串 hello message 并回车
Step5:启动消费者
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testtopic --from-beginning hello message
此时在发送消息侧不断输入消息并回车,消费者会不断打印收到的消息,至此Kafka的简单收发消息运行完毕
RocketMQ的部署与使用
我们依然根据官网文档中的 Quick Start节的提示开始操作,官网建议使用64位Linux/Unix/Mac操作系统,我们使用的是64位的Centos 6.x,建议使用JDK1.8+,我们上面部署Kafka的时候已经安装了jdk1.8,需要maven3.2.x,我的系统上没有,所以首先需要安装maven:
安装maven:
1. 下载maven安装包,我这里使用的是更高一点的版本:
wget http://mirrors.cnnic.cn/apache/maven/maven-3/3.5.4/binaries/apache-maven-3.5.4-bin.tar.gz
2. 解压maven安装包:
tar -zxvf apache-maven-3.5.4-bin.tar.gz
3. 配置maven到profile文件中以便启动的时候自动设置环境变量 vim /etc/profile 添加下面两行:
export MAVEN_HOME=/data/rocket_mq/apache-maven-3.5.4 export PATH=$MAVEN_HOME/bin:$PATH
注意这里你应该换成你自己的对应的目录
4. 使环境变量生效:
source /etc/profile
5. 检验是否正常:
mvn -version Apache Maven 3.5.4 (1edded0938998edf8bf061f1ceb3cfdeccf443fe; 2018-06-18T02:33:14+08:00) Maven home: /data/rocket_mq/apache-maven-3.5.4 Java version: 1.8.0_212, vendor: Oracle Corporation, runtime: /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.212.b04-0.el6_10.x86_64/jre Default locale: en_US, platform encoding: UTF-8 OS name: "linux", version: "4.1.0-32.el6.ucloud.x86_64", arch: "amd64", family: "unix"
下载RocketMQ的源码并编译:
wget https://archive.apache.org/dist/rocketmq/4.7.0/rocketmq-all-4.7.0-source-release.zip unzip rocketmq-all-4.7.0-source-release.zip cd rocketmq-all-4.7.0/ mvn -Prelease-all -DskipTests clean install -U cd distribution/target/rocketmq-4.7.0/rocketmq-4.7.0
编译成功之后的截图如下:
运行nameserver服务:
sh bin/mqnamesrv ERROR: Please set the JAVA_HOME variable in your environment, We need java(x64)! !!
首先我们在编译RocketMQ的时候我们已经将目录定位到 cd distribution/target/rocketmq-4.7.0/rocketmq-4.7.0 在此目录下运行nameserver服务报错,根据提示我们需要设置 JAVA_HOME 环境变量,如下(修改为你自己的地址):
export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.212.b04-0.el6_10.x86_64/ export CLASSPATH=.:$JAVA_HOME/jre/lib/rt.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar export PATH=$PATH:$JAVA_HOME/bin
之后再次运行:
sh bin/mqnamesrv OpenJDK 64-Bit Server VM warning: Using the DefNew young collector with the CMS collector is deprecated and will likely be removed in a future release OpenJDK 64-Bit Server VM warning: UseCMSCompactAtFullCollection is deprecated and will likely be removed in a future release. The Name Server boot success. serializeType=JSON
可以看到Name Server运行成功,这里提一下,Name Server是为RocketMQ提供名字服务的模块,其角色与zookeeper在Kafka中的作用类似
启动Broker服务:
在启动Broker这里有一点要注意的是,我们这里运行命令所在的目录跟上面运行Name Server的目录一样,都是在 distribution/target/rocketmq-4.7.0/rocketmq-4.7.0 这个目录下运行,并且上面需要导入的 JAVA_HOME 等环境变量这里都需要,除此之外,运行Broker我们还需要修改一下配置文件,在当前目录下(distribution/target/rocketmq-4.7.0/rocketmq-4.7.0) 修改文件 vim conf/broker.conf 在文件后面添加如下一行配置:
brokerIP1 = 127.0.0.1
设置Broker的ip,如果不设置的话,当机器存在多网卡的时候,例如一个内外ip和一个外网ip的情况,Broker可能会配置第一个ip地址,导致我们后面的生产者启动的时候写入消息失败报告下面的错误:
为了测试方便,我们直接设置BrokerIP1为本机回环ip地址127.0.0.1,然后通过配置文件启动Broker:
bin/mqbroker -n localhost:9876 -c conf/broker.conf The broker[broker-a, 127.0.0.1:10911] boot success. serializeType=JSON and name server is localhost:9876
启动生产者发送消息:
Broker启动完成之后,我们接下来启动生产者发送消息(同样要定位到 distribution/target/rocketmq-4.7.0/rocketmq-4.7.0 并导入 JAVA_HOME 等环境变量):
export NAMESRV_ADDR=localhost:9876 sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
03:08:05.769 [main] DEBUG i.n.u.i.l.InternalLoggerFactory – Using SLF4J as the default logging framework
RocketMQLog:WARN No appenders could be found for logger (io.netty.util.internal.PlatformDependent0).
RocketMQLog:WARN Please initialize the logger system properly.
SendResult [sendStatus=SEND_OK, msgId=20020AB3FD9D000000000000000000017E931540E19D439EE2B00001, offsetMsgId=7F00000100002A9F00000000000639AE, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=1], queueOffset=500]
……
03:08:07.668 [NettyClientSelector_1] INFO RocketmqRemoting – closeChannel: close the connection to remote address[127.0.0.1:9876] result: true
03:08:07.669 [NettyClientSelector_1] INFO RocketmqRemoting – closeChannel: close the connection to remote address[127.0.0.1:10911] result: true
可以看到中间我省略了很多日志,这些日志表示生产者向服务器写入了大量的消息,由于输出太多,我只列出2条
启动消费者消费消息:
接着我们启动消费者消费消息(同样要定位到 distribution/target/rocketmq-4.7.0/rocketmq-4.7.0 并导入 JAVA_HOME 等环境变量):
export NAMESRV_ADDR=localhost:9876 sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
03:11:37.229 [main] DEBUG i.n.u.i.l.InternalLoggerFactory – Using SLF4J as the default logging framework
Consumer Started.
ConsumeMessageThread_1 Receive New Messages: [MessageExt [brokerName=broker-a, queueId=3, storeSize=202, queueOffset=500, sysFlag=0, bornTimestamp=1586804886200, bornHost=/127.0.0.1:42523, storeTimestamp=1586804886203, storeHost=/127.0.0.1:10911, msgId=7F00000100002A9F0000000000063B42, commitLogOffset=408386, bodyCRC=1032136437, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic=\’TopicTest\’, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=750, CONSUME_START_TIME=1586805097731, UNIQ_KEY=20020AB3FD9D000000000000000000017E931540E19D439EE2B80003, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 51], transactionId=\’null\’}]]
同样是大量消费消息的日志,这里也只列出前面2条输出,至此,RocketMQ的部署与简单消息收发使用完毕
RabbitMQ的部署与使用
RabbitMQ的部署我们同样参照官网给出的步骤,找到官网的 Get Started 点开按钮 “Download + Installation” ,RabbitMQ有多种运行方式,可以基于docker镜像运行,我们本次不使用docker这种方式部署。我们使用常规的部署方法,RabbitMQ使用Erlang语言编写,运行需要Erlang的运行时环境,所以我们首先需要安装Erlang环境,RabbitMQ团队为我们提供了一个Erlang安装包,其仅包含运行RabbitMQ所需要的全部组件,为了方便我们就使用此Erlang安装包,安装起来很容易:
安装Erlang环境:
wget https://github.com/rabbitmq/erlang-rpm/releases/download/v22.3.1/erlang-22.3.1-1.el6.x86_64.rpm yum install erlang-22.3.1-1.el6.x86_64.rpm
没有报错,我这里是下载的Centos版本的二进制安装包,其他系统可以自行去RabbitMQ官网下载,Erlang安装好之后,我们就可以安装RabbitMQ服务了:
安装RabbitMq:
wget https://github.com/rabbitmq/signing-keys/releases/download/2.0/rabbitmq-release-signing-key.asc rpm --import rabbitmq-release-signing-key.asc wget http://wangqiguoy.cn-bj.ufileos.com/rabbitmq-server-3.8.3-1.el6.noarch.rpm yum install rabbitmq-server-3.8.3-1.el6.noarch.rpm
我安装的时候没有任何报错,直接安装完成,比Kafka和RocketMQ要顺利多了,RabbitMQ提供了一个web界面可以对其进行管理,可以在上面看到当前的队列,通道信息,消息堆积情况等等,通过下面的命令开启插件:
开启web管理插件:
rabbitmq-plugins enable rabbitmq_management Enabling plugins on node rabbit@10-179-253-157: rabbitmq_management The following plugins have been configured: rabbitmq_management rabbitmq_management_agent rabbitmq_web_dispatch Applying plugin configuration to rabbit@10-179-253-157... The following plugins have been enabled: rabbitmq_management rabbitmq_management_agent rabbitmq_web_dispatch started 3 plugins.
开启成功之后,可以通过一个http的地址访问web界面管理RabbitMQ,地址形式是:http://ip:15672/ 将ip地址换成自己的服务器ip地址即可,登陆界面如下:
修改RabbitMQ的配置文件以登陆其web界面:
vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.3/ebin/rabbit.app {loopback_users, [<<"guest">>]}, 改成 {loopback_users, [guest]}
修改好之后,重启服务生效,可以通过上面的web界面登录,用户名和密码都是guest,登陆进去之后的界面如下图:
功能还是很丰富的,可以查看connection、channel、exchange、queue等信息,还可以管理用户权限,除了通过web界面管理RabbitMQ之外,Rabbit还提供一个命令行工具rabbitmqctl来管理RabbitMQ,该工具的具体使用可以查看其官网文档,另外对RabbitMQ常用的启动停止操作可以参考如下命令:
service rabbitmq-server start service rabbitmq-server stop service rabbitmq-server restart service rabbitmq-server status
发送消息与消费消息:
RabbitMQ没有提供命令行的测试方式,不过RabbitMQ提供了很多客户端的demo可供使用,这里我选择的是使用nodejs,首先要准备node的环境如下:
yum install nodejs yum install npm npm install amqplib
最后一步安装 amqplib 即是nodejs中要使用的与RabbitMQ相关的客户端库,我们的生产者 send.js 代码如下:
//send.js var amqp = require(\'amqplib/callback_api\'); amqp.connect(\'amqp://localhost\', function(error0, connection) { if (error0) { throw error0; } connection.createChannel(function(error1, channel) { if (error1) { throw error1; } var queue = \'hello\'; var msg = \'{"name":"Hello World!"}\'; channel.assertQueue(queue, { durable: false }); channel.sendToQueue(queue, new Buffer(msg)); console.log(" [x] Sent %s", msg); }); setTimeout(function() { connection.close(); process.exit(0); }, 500); });
消费者 receive.js 的代码如下:
//receive.js var amqp = require(\'amqplib/callback_api\'); amqp.connect(\'amqp://localhost\', function(error0, connection) { if (error0) { throw error0; } connection.createChannel(function(error1, channel) { if (error1) { throw error1; } var queue = \'hello\'; channel.assertQueue(queue, { durable: false }); console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", queue); channel.consume(queue, function(msg) { console.log(" [x] Received %s", msg.content.toString()); }, { noAck: true }); }); });
分别运行 send.js 与 receive.js 得到如下输出:
node send.js [x] Sent {"name":"Hello World!"} node receive.js [*] Waiting for messages in hello. To exit press CTRL+C [x] Received {"name":"Hello World!"}
至此RabbitMQ的部署以及简单消息收发使用完成