一、MQ简单介绍
我们在做项目的时候经常会使用消息队列（MQ）实现不同的系统或子应用等之间的数据的传递，接下来简单的介绍几款流行的MQ应用，想知道详细且具体的MQ知识可以自行度娘或Google。这篇文章就讲的很不错https://www.cnblogs.com/stone531/p/10519279.html。
1.ZeroMQ
号称最快的消息队列系统，尤其针对大吞吐量的需求场景。扩展性好，开发比较灵活，采用C语言实现，实际上只是一个socket库的重新封装，如果做为消息队列使用，需要开发大量的代码。ZeroMQ仅提供非持久性的队列，也就是说如果down机，数据将会丢失。其中，Twitter的Storm中使用ZeroMQ作为数据流的传输。
2.RabbitMQ
结合erlang语言本身的并发优势，支持很多的协议：AMQP，XMPP, SMTP, STOMP，也正是如此，使的它变的非常重量级，更适合于企业级的开发。性能较好，但是不利于做二次开发和维护。
3.ActiveMQ
历史悠久的开源项目，是Apache下的一个子项目。已经在很多产品中得到应用，实现了JMS1.1规范，可以和spring-jms轻松融合，实现了多种协议，不够轻巧（源代码比RocketMQ多），支持持久化到数据库，对队列数较多的情况支持不好。
4.Redis
做为一个基于内存的K-V数据库，其提供了消息订阅的服务，可以当作MQ来使用，目前应用案例较少，且不方便扩展。入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如 果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于 Redis。
5.Kafka/Jafka
Kafka是Apache下的一个子项目，是一个高性能跨语言分布式发布/订阅消息队列系统，而Jafka是在Kafka之上孵化而来的，即Kafka的一个升级版。
具有以下特性：
①：快速持久化，可以在O(1)的系统开销下进行消息持久化；
②：高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；完全的分布式系统，Broker、Producer、Consumer都原生自③：动支持分布式，自动实现负载均衡；
④：支持Hadoop数据并行加载，对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。
⑤：Kafka通过Hadoop的并行加载机制统一了在线和离线的消息处理。Apache Kafka相对于ActiveMQ是一个非常轻量级的消息系统，除了性能非常好之外，还是一个工作良好的分布式系统。
二、安装java环境
今天我就来详细介绍一下单点在虚拟机中部署一个kafka消息队列，并用python简单的实现消息的生产与消费。
1.先在虚拟机安装java环境，下载链接 (http://www.oracle.com/technetwork/java/javase/downloads/index.html)，可以根据自己的系统进行下载。
2、进行解压
sudo tar -zxvf jdk-8u171-linux-x64.tar.gz
则解压到了当前目录下，解压后可以把解压文件移动到自己想要放的目录下，使用mv命令
sudo mv jdk1.8.0_171 /usr/lib/xxx
3、进行配置
使用全局设置方法，它是所有用户的共用的环境变量
命令如下：
$sudo vim ~/.bashrc
然后把如下命令复制到最底部
export JAVA_HOME=/usr/local/java/jdk1.8.0_25
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH
export JAVA_HOME=后面要填写自己解压后的jdk的路径
4、生效~/.bashrc文件
命令如下：
$sudo source ~/.bashrc
5、测试是否安装成功
java -version
查看版本号是否改变
三、安装zookeeper
1.下载zookeeper.
链接 https://zookeeper.apache.org/releases.html#download
2.下载解压后，进入解压目录，从 conf 目录下拷贝 zoo_sample.cfg 到该目录下并重命名为 zoo.cfg。
3.配置 zoo.cfg ，指定数据目录和日志目录
4.启动zookeeper,进入bin目录下使用命令./zkServer.sh start
./zkServer.sh start
如果shell终端出现以下打应，说明启动成功。
ZooKeeper JMX enabled by default
Using config: /home/wangping/桌面/zookeeper-3.4.14/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
5.查看zookeeper状态。进入bin目录下使用命令./zkServer.sh status
./zkServer.sh status
如果shell终端出现以下打应，说明zookeeper单点模式启动成功。
ZooKeeper JMX enabled by default
Using config: /home/wangping/桌面/zookeeper-3.4.14/bin/../conf/zoo.cfg
Mode: standalone
6.集群配置，需要在zoo.cfg配置文件中添加如下配置。并在zoo.cfg中配置的日志文件目录中同级创建myid文件，第一台机器里面写1，以此类推。
四、安装kafka
1.先下载kafka，下载链接(http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.4.0/kafka_2.13-2.4.0.tgz)，解压。
2.在config文件夹下配置server.properties。默认配置 advertised.listeners=PLAINTEXT://:your.host.name:9092 修改为 advertised.listeners=PLAINTEXT://:127.0.0.1:9092
3.启动kafka。
bin/kafka-server-start.sh config/server.properties
4.创建topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
其中test为创建的topic名称。
如上图，创建了一个名为 test 的主题，其中包含一个分区和一个副本因子。 创建成功之后会输出： Created topic "test".创建主题后，系统会在config / server.properties文件中的"/ tmp / kafka-logs /"中指定的创建主题的日志。
5.其他kafka基本命令。
bin/kafka-topics.sh --list --zookeeper localhost:2181 查询topic列表
bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic test 查看topic信息
bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic demo 删除topic
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test 启动生产者
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning 启动消费者
五、python通过kafka简单实现生产者与消费者。
1.创建生产者
from kafka import KafkaProducer
kafka_host = '127.0.0.1' # kafka服务器地址
kafka_port = 9092 # kafka服务器的端口
producer = KafkaProducer(bootstrap_servers=['{kafka_host}:{kafka_port}'.format(
kafka_host = kafka_host,
kafka_port = kafka_port
)])
#调用send方法，发送的消息为hi
producer.send('test', key="hello".encode('utf-8'), value= "hi".encode('utf-8'))
#发送信息需要编码加上encode('utf-8')，否则会报错AssertionError: xxx must be bytes
producer.flush()
2.创建消费者
from kafka import KafkaConsumer
kafka_host = '127.0.0.1' # kafka服务器地址
kafka_port = 9092 # kafka服务器的端口
#消费topic=>test里的消息，并指定group_id(自定义)，多个机器或进程想顺序消费，可以指定同一个group_id，
# 如果想一条消费多次消费，可以换一个group_id,会从头开始消费
consumer = KafkaConsumer(
'test',
group_id = 'my-group',
bootstrap_servers = ['{kafka_host}:{kafka_port}'.format(kafka_host=kafka_host, kafka_port=kafka_port)]
)
for message in consumer:
print("message is=>%s"%message.value)
简单的利用python实现对消息队列的使用就完成了。祝大家新年快乐。
