RabbitMQ模型
| 名词 | 描述 |
| Queue | 用于存储消息，消费者直接绑定Queue进行消费消息 |
| Exchange | 生产者将消息发送到Exchange，由交换器将消息通过匹配Exchange Type、Binding Key、Routing Key后路由到一个或者多个队列中。 |
| Exchange Type | Direct、Fanout、Topic、Headers |
| Routing Key | 生产者发送消息给Exchange会指定一个Routing Key。 |
| Binding Key | 在绑定Exchange与Queue时会指定一个Binding Key |
1.Exchange在声明时会绑定Queue和Binding Key，当Exchange收到消息会根据消息的
2.Routing Key与Exchange Type、Binding Key进行匹配，最后会路由到相关的队列当中。
Fanout，将消息发送到与该交换器所绑定的所有队列中，与Routing Key、Bind Key无关，这就是广播模式。
Topic，通过对消息的Routing Key和Exchange、Queue进行匹配，将消息路由给一个或多个队列，以此来达到发布/订阅模式。
Direct，把消息路由到哪些Bind Key和Routing Key完全匹配的队列中。
Headers，不依赖与路由键的匹配规则，基本用不上。
3.消费者会直接订阅Queue里的消息进行消费，多个消费者订阅同个Queue会形成消息竞争状态，以此达到负载均衡作用。
Kafka模型
| 名词 | 描述 |
| Topic | 队列是通过Topic进行隔离的，生产者发送消息必须指定Topic |
| Broker | 一个Kafka Server的被称为一个Broker。 |
| Partition | 每个Topic可以包含多个Partition，多个Partition会平均分配给同一个Consumer Group里的不同Consumer进行消费 |
| Consumer Group | 不在同一个Group 的Consumer能重复消费同一条消息（订阅），相同Group的Consumer存在消费竞争（负载均衡） |
- Kafka与RabbitMQ比没有Exchange的概念，生产者直接发消息Topic（队列）。
- Kafka的订阅者是通过消费组（Consumer Group）来体现的，每个消费组都可以重复消费Topic一份完整的消息，不同消费组之间消费进度彼此不受影响。例如Message1能被Consumer Group 1和Consumer Group2里的消费者都消费一次。
- 消费组中包含多个消费者，同个Group的消费者之间是竞争消费的关系。例如Message2只能够被Consumer Group里某一个Consumer只消费一次。
- Kafka具有消息存储的功能，消息被消费后不会被立即删除，因为需要被不同的Consumer Group多次消费同个消息，因此会在Topic维护一个Consumer Offset，每消费成功Offset自增1.
功能对比
| 对比项 | RabbitMQ | Kafka |
| 吞吐量 | 低 | 高 |
| 有序性 | 全局有序性 | 分区有序性 |
| 消息可靠性 | 多策略组合 | 消息持久化 |
| 流处理 | 不支持 | 支持 |
| 时效性 | 高 | 中 |
| 运维便捷度 | 高 | 中 |
| 系统依赖 | 无 | zookeeper |
| Web监控 | 自带 | 第三方 |
| 优先级队列 | 支持 | 不支持 |
| 死信 | 支持 | 不支持 |
| 客户端支持 | 支持多种语言 | |
| 社区生态 | 好 | |
| 安全机制 | （TLS/SSL、SASL）身份认证和（读写）权限控制 | |
| 消息回溯 | 支持 | 不支持 |
对比描述
共同点
RabbitMQ与Kafka都有很好的客户端语言支持、安全机制与生态支持。
性能
Kafka的诞生的是处理高并发日志的，吞吐量比较高，每秒请求数达到数十万量级
RabbitMQ每秒请求数则为万级别，有测试报告指出Kafka是RabbitMQ的10倍以上性能。
运维便捷
RabbitMQ相对比较方便，可以使用yum或者docker安装，自带Web管理UI，没有额外的依赖，除了需要做镜像队列外需要引入HAproxy。
Kafka则需要依赖Zookeeper，也没有自带的管理工具，可以使用第三方的Kafka Eagle代替，Kafka Manager过于难用，另外Kafka没有yum安装，docker镜像也是社区人员自己建的。
有序性
RabbitMQ理论上是全局有序性的，但是由于【发后既忘】+【自动确认】机制的原因，如果在同个队列的多个消费者做相同的业务处理时，他们的各自的执行任务无法保证有序完成。如果确保100%有序可以使用【非自动确认】，但会影响消费性能。
Kafka支持分区有序性，如果对有序性有严格要求可以设置单个Partition，可是单个Partition并发性比较低，因此在多个Partition情况下可以根据业务指定key把相关的消息路由到同一个Partition，例如相同UserId行为信息可以到Partition 1进行处理。
时效性
Kafka基本上无论在客户端还是服务端都是以【异步批量】的机制进行处理，通俗的讲就是先攒起来一堆消息，到了某个阀值再发送，也会导致一些消息可靠性与消息有时效上的问题，当然可以通过各种配置策略进行解决。
消息回溯
Kafka在消费完了消息后不会立即删除，只会修改offset，如果之前部分业务消费失败了可以重新设置offset进行重新消费。
RabbitMQ则是[发后既忘]的机制，一但消费者确认消息则删除，但是可以通过死信进行补偿消费。此外RabbitMQ在队列消息堆积多的情况下性能表现不佳，所以尽可能的及时消费消息。
特色功能
RabbitMQ具有死信的功能，可以通过死信形成重复消费与延时发送。
Kafka具有流处理功能，可以收集用户的行为日志进行存储与分析。
Kafka为什么快？
关键核心技术点：
- 异步批量处理
- 磁盘顺序读写
- 操作系统PageCache缓存数据
- 零拷贝加速消费
Kafka的诞生就是为了高并发日志处理的，那么在他整个机制里使用了很多批量、异步、缓存。例如生产者客户端，他会积累一定量（条数、大小）的消息，再批量的发给kafka broker，如果在这段时间客户端服务挂了，就等于消息丢失了。当broker接受到了消息后，还有一堆骚操作-异步刷盘，也就是生产者发送给broker之后他是记录在缓存的，每隔一段时间才会持久化到磁盘，假如这段真空期broker挂了，消息也是丢了。
Kafka是否消息不可靠？
Kafka快是因为牺牲了消息可靠换取回来的性能，在最早期版本的确没提供消息可靠的策略，经过多个版本迭代后的功能完善，已经不存在这种旧观念。那么可靠的关键点有以下：
生产者
设置ack：
- 0：producer不等待broker的ack，broker一接收到还没有写入磁盘就已经返回，可靠性最低；
- 1：producer等待broker的ack，partition的leader刷盘成功后返回ack，如果在follower同步成功之前leader故障，那么将会丢失数据，可靠性中；
- -1：producer等待broker的ack，partition的leader和follower全部落盘成功后才返回ack，数据一般不会丢失，延迟时间长但是可靠性高
消费者
设置enable.auto.commitrue，不管执行结果如何，消费者会自动提交offset。
选型总结
对于选择Kafka还是RabbitMQ，主要考虑三个因素：吞吐量、运维能力和平台熟悉度。如果是需要流处理和高并发的日志处理，首选Kafka。但是大部分公司并没有什么高并发的处理，因此可以着重考虑运维程度和平台熟悉度，前面提到Kafka也是有策略可以设置消息可靠的。RabbiMQ运维比较直接，包括客户端EasyNetQ使用简易性，基本上就是”开箱即用“。
客户端
| MQ | 客户端名称 | 文档地址 |
| Kafka | confluent-kafka-dotnet | https://github.com/confluentinc/confluent-kafka-dotnet/wiki |
| RabbitMQ | EasyNetQ | https://github.com/EasyNetQ/EasyNetQ/wiki/Introduction |
RabbitMQ单节点部署
安装
yum install -y rabbitmq-server
开放相关端口
firewall-cmd --permanent --add-port=15672/tcp firewall-cmd --permanent --add-port=5672/tcp firewall-cmd --reload
启动服务
service rabbitmq-server start
设置开机启动
chkconfig rabbitmq-server on
启动web管理界面
rabbitmq-plugins enable rabbitmq_management
增加访问admin用户，默认用户guest只能本地访问。
rabbitmqctl add_user admin 123456
设置admin用户为管理员角色
rabbitmqctl set_user_tags admin administrator
设置默认admin用户访问权限
rabbitmqctl set_permissions -p "/" admin "." "." ".*"
重启服务
service rabbitmq-server restart
浏览器访问：http://IP:15672
RabbitMQ镜像集群部署
1.所有实例都需要执行以上单实例部署的命令
2.所有实例服务器的打开相关端口
firewall-cmd --permanent --add-port=59984/tcp firewall-cmd --permanent --add-port=25672/tcp firewall-cmd --permanent --add-port=4369/tcp firewall-cmd --reload
3.在每台服务器添加其他rabbitmq 实例服务的hosts映射
#修复服务器hosts文件 vim /etc/hosts 192.168.88.138 server-a 192.168.88.139 server-c
先查看集群信息
rabbitmqctl cluster_status
4.同步erlang cookie
在server-c（192.168.88.139）执行远程拷贝到server-a(192.168.88.138)
rabbitmqctl stop scp /var/lib/rabbitmq/.erlang.cookie root@192.168.88.138:/var/lib/rabbitmq/.erlang.cookie rabbitmq-server -detached
5.server-c加入RabbitMQ集群
rabbitmqctl stop_app rabbitmqctl join_cluster rabbit@server-a rabbitmqctl start_app
6.设置镜像队列策略
#任意节点执行
rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'
7.安装HAProxy并启动
在服务器server-b执行下面命令：
yum install haproxy -y service haproxy start chkconfig haproxy on vim /etc/haproxy/haproxy.cfg
覆盖配置
# Global settings
#---------------------------------------------------------------------
global
# to have these messages end up in /var/log/haproxy.log you will
# need to:
#
# 1) configure syslog to accept network log events. This is done
# by adding the '-r' option to the SYSLOGD_OPTIONS in
# /etc/sysconfig/syslog
#
# 2) configure local2 events to go to the /var/log/haproxy.log
# file. A line like the following can be added to
# /etc/sysconfig/syslog
#
# local2.* /var/log/haproxy.log
#
log 127.0.0.1 local2
chroot /var/lib/haproxy
pidfile /var/run/haproxy.pid
maxconn 4000
user haproxy
group haproxy
daemon
# turn on stats unix socket
stats socket /var/lib/haproxy/stats
#---------------------------------------------------------------------
# common defaults that all the 'listen' and 'backend' sections will
# use if not designated in their block
#---------------------------------------------------------------------
defaults
mode http
log global
option httplog
option dontlognull
option http-server-close
option forwardfor except 127.0.0.0/8
option redispatch
retries 3
timeout http-request 10s
timeout queue 1m
timeout connect 10s
timeout client 1m
timeout server 1m
timeout http-keep-alive 10s
timeout check 10s
maxconn 3000
#---------------------------------------------------------------------
# main frontend which proxys to the backends
#---------------------------------------------------------------------
###############RabbitMQ服务#################
listen rabbitmq_cluster
bind 0.0.0.0:5672
mode tcp
balance roundrobin
server server-a 192.168.88.138:5672 check inter 5000 rise 2 fall 3 weight 1
server server-c 192.168.88.139:5672 check inter 5000 rise 2 fall 3 weight 1
###############RabbitMQ管理界面#################
listen rabbitmq_ui
bind 0.0.0.0:15672
server server-a 192.168.88.138:15672
server server-c 192.168.88.139:15672
开放端口
firewall-cmd --permanent --add-port=15672/tcp firewall-cmd --permanent --add-port=5672/tcp firewall-cmd --reload
Kafka单节点部署
Zookeeper部署
下载Zookeeper并启动
docker run -d --restart always --name zookeeper -p 2181:2181 -v /root/zookeeper/data:/data -v /root/zookeeper/conf:/conf -v /root/zookeeper/logs:/logs zookeeper:3.6.1
开放2181端口
firewall-cmd --permanent --add-port=2181/tcp firewall-cmd --reload
Kafka服务部署
下载kafka 镜像并启动
docker run -d --name kafka -p 9092:9092 -e KAFKA_BROKER_ID=1 -e KAFKA_ZOOKEEPER_CONNECT=192.168.88.139:2181 -e KAFKA_ADVERTISED_HOST_NAME=192.168.88.141 -e KAFKA_ADVERTISED_PORT=9092 wurstmeister/kafka:2.12-2.5.0
创建目录并拷贝
mkdir /root/kafka docker cp kafka:/opt/kafka/config /root/kafka/config
删除原有的容器并重新创建
docker stop kafka docker rm kafka docker run -d --name kafka -p 9092:9092 -e KAFKA_BROKER_ID=1 -e KAFKA_ZOOKEEPER_CONNECT=192.168.88.139:2181 -e KAFKA_ADVERTISED_HOST_NAME=192.168.88.141 -e KAFKA_ADVERTISED_PORT=9092 -v /root/kafka/config: /opt/kafka/config wurstmeister/kafka:2.12-2.5.0
开放9092端口
firewall-cmd --permanent --add-port=9092/tcp firewall-cmd --reload
Kafka-eagle
下载jdk依赖
yum -y install java-1.8.0-openjdk*
下载kafka-eagle-bin包
wget -o kafka-eagle-bin.tar.gz https://codeload.github.com/smartloli/kafka-eagle-bin/tar.gz/v2.0.1
解压
tar -zxvf kafka-eagle-bin.tar.gz tar -zxvf kafka-eagle-bin-2.0.1/kafka-eagle-web-2.0.1-bin.tar.gz mv kafka-eagle-web-2.0.1 kafka-eagle
添加环境变量
vim /etc/profile export JAVA_HOME=/usr export KE_HOME=/etc/kafka-eagle export PATH=$PATH:$KE_HOME/bin:$JAVA_HOME/bin
生效环境变量
source /etc/profile
修改Kafka-eagle配置
cd /etc/kafka-eagle/conf vim system-config.properties #注释 #cluster2.zk.list=xdn10:2181,xdn11:2181,xdn12:2181 #cluster2.kafka.eagle.offset.storage=zk #cluster1.zk.acl.enable=false #cluster1.zk.acl.schema=digest #cluster1.zk.acl.username=test #cluster1.zk.acl.password=test123 修改 kafka.eagle.zk.cluster.alias=cluster1 cluster1.zk.list=192.168.88.139:2181 kafka.eagle.metrics.charts=true kafka.eagle.driver=org.sqlite.JDBC kafka.eagle.url=jdbc:sqlite:/etc/kafka-eagle/db/ke.db kafka.eagle.username=root kafka.eagle.password=root
启动kafka-eagle服务
cd /etc/kafka-eagle/bin chmod +x ke.sh ke.sh start
开启防火墙
firewall-cmd --permanent --add-port=8048/tcp firewall-cmd --reload
浏览器访问：http://IP:8048
