RabbitMQ消息中间件知识汇总

Alan

5 年前

其它主流消息中间件：AcitveMQ(老牌，但性能略差)、Kafka（性能好，可靠性略差）、RocketMQ（收费）

RabbitMQ基于 AMQP 协议实现。适用于对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求次之。

RabbitMQ 整体架构

安装

官网地址：https://www.rabbitmq.com/

Erlang 与 RabbitMQ 对应匹配版本：https://www.rabbitmq.com/which-erlang.html

安装环境：CentOS 7

安装 Erlang

方法一

参考地址：https://github.com/rabbitmq/erlang-rpm

# sudo vi /etc/yum.repos.d/rabbitmq_erlang.repo
[rabbitmq_erlang]
name=rabbitmq_erlang
baseurl=https://packagecloud.io/rabbitmq/erlang/el/7/$basearch
repo_gpgcheck=1
gpgcheck=0
enabled=1
gpgkey=https://packagecloud.io/rabbitmq/erlang/gpgkey
sslverify=1
sslcacert=/etc/pki/tls/certs/ca-bundle.crt
metadata_expire=300

[rabbitmq_erlang-source]
name=rabbitmq_erlang-source
baseurl=https://packagecloud.io/rabbitmq/erlang/el/7/SRPMS
repo_gpgcheck=1
gpgcheck=0
enabled=1
gpgkey=https://packagecloud.io/rabbitmq/erlang/gpgkey
sslverify=1
sslcacert=/etc/pki/tls/certs/ca-bundle.crt
metadata_expire=300

# 安装
sudo yum install erlang -y

# sudo vi /etc/yum.repos.d/rabbitmq_erlang.repo

[rabbitmq_erlang]

name=rabbitmq_erlang

baseurl=https://packagecloud.io/rabbitmq/erlang/el/7/$basearch

repo_gpgcheck=1

gpgcheck=0

enabled=1

gpgkey=https://packagecloud.io/rabbitmq/erlang/gpgkey

sslverify=1

sslcacert=/etc/pki/tls/certs/ca-bundle.crt

metadata_expire=300

[rabbitmq_erlang-source]

name=rabbitmq_erlang-source

baseurl=https://packagecloud.io/rabbitmq/erlang/el/7/SRPMS

repo_gpgcheck=1

gpgcheck=0

enabled=1

gpgkey=https://packagecloud.io/rabbitmq/erlang/gpgkey

sslverify=1

sslcacert=/etc/pki/tls/certs/ca-bundle.crt

metadata_expire=300

# 安装

sudo yum install erlang -y

方法二

下载地址：https://www.erlang-solutions.com/resources/download.html

# 安装依赖
sudo yum -y install epel-release        
sudo yum -y install unixODBC unixODBC-devel wxBase wxGTK SDL wxGTK-gl

wget https://packages.erlang-solutions.com/erlang/rpm/centos/7/x86_64/esl-erlang_20.3.8.22-1~centos~7_amd64.rpm
rpm -ivh esl-erlang_20.3.8.22-1~centos~7_amd64.rpm

# 安装依赖

sudo yum -y install epel-release

sudo yum -y install unixODBC unixODBC-devel wxBase wxGTK SDL wxGTK-gl

wget https://packages.erlang-solutions.com/erlang/rpm/centos/7/x86_64/esl-erlang_20.3.8.22-1~centos~7_amd64.rpm

rpm -ivh esl-erlang_20.3.8.22-1~centos~7_amd64.rpm

可通过 erl 进入命令行查看

安装 RabbitMQ

以3.6.16为例

sudo yum install socat -y # RabbitMQ 需依赖 socat
wget https://github.com/rabbitmq/rabbitmq-server/releases/download/rabbitmq_v3_6_16/rabbitmq-server-3.6.16-1.el7.noarch.rpm
sudo rpm -ivh rabbitmq-server-3.6.16-1.el7.noarch.rpm

sudo yum install socat -y # RabbitMQ 需依赖 socat

wget https://github.com/rabbitmq/rabbitmq-server/releases/download/rabbitmq_v3_6_16/rabbitmq-server-3.6.16-1.el7.noarch.rpm

sudo rpm -ivh rabbitmq-server-3.6.16-1.el7.noarch.rpm

注意：如不安装 socat，会报如下错误

error: Failed dependencies:
	socat is needed by rabbitmq-server-3.6.16-1.el7.noarch

1 2	error: Failed dependencies: socat is needed by rabbitmq-server-3.6.16-1.el7.noarch

配置文件

# sudo vi /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.16/ebin/rabbit.app
#{loopback_users, [<<"guest">>]},修改为
{loopback_users, [guest]},

# sudo vi /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.16/ebin/rabbit.app

#{loopback_users, [<<"guest">>]},修改为

{loopback_users, [guest]},

服务启动/停止

# 启动服务
sudo rabbitmq-server start& # 或者sudo rabbitmqctl start_app
# 验证服务启动：sudo lsof -i:5672
# 停止服务
sudo rabbitmqctl stop_app # 或sudo rabbitmq-server stop
# 管理插件
sudo rabbitmq-plugins enable rabbitmq_management # 查看sudo rabbitmq-plugins list

# 启动服务

sudo rabbitmq-server start& # 或者sudo rabbitmqctl start_app

# 验证服务启动：sudo lsof -i:5672

# 停止服务

sudo rabbitmqctl stop_app # 或sudo rabbitmq-server stop

# 管理插件

sudo rabbitmq-plugins enable rabbitmq_management # 查看sudo rabbitmq-plugins list

启用插件后在浏览器中访问http://your.ip.addr:15672/，若无法访问请注意检查防火墙相关配置，默认登录用户名/密码： guest/guest

Docker 安装

docker run -dit --name some-rabbit --restart=always -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 5672:5672 rabbitmq:management

1	docker run -dit --name some-rabbit --restart=always -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 5672:5672 rabbitmq:management

命令行与管控台

# 以下 vhostpath 需替换为虚拟主机路径通过 list_vhosts 命令可进行查看
sudo rabbitmqctl status # 节点状态
sudo rabbitmqctl add_user username password # 添加用户
sudo rabbitmqctl list_users  # 列出所有用户
sudo rabbitmqctl delete_user username  # 删除用户
sudo rabbitmqctl clear_permissions -p vhostpath username  # 清除用户权限
sudo rabbitmqctl list_user_permissions username  # 列出用户权限
sudo rabbitmqctl change_password username newpassword  # 修改密码
sudo rabbitmqctl set_permissions -p vhostpath username ".*" ".*" ".*"  # 设置用户权限 

sudo rabbitmqctl add_vhost vhostpath  # 创建虚拟主机
sudo rabbitmqctl list_vhosts  # 列出所有虚拟主机
sudo rabbitmqctl list_permissions -p vhostpath # 列出虚拟主机上所有权限
sudo rabbitmqctl delete_vhost vhostpath  # 删除虚拟主机

sudo rabbitmqctl list_queues  # 查看所有队列信息
sudo rabbitmqctl -p vhostpath purge_queue blue  # 清除队列里的消息

sudo rabbitmqctl reset  # 移除所有数据，要在 rabbitmqctl stop_app之后使用
sudo rabbitmqctl join_cluster <clusternode> [--ram|disc]  # 组成集群命令
sudo rabbitmqctl cluster_status  # 查看集群状态
sudo rabbitmqctl change_cluster_node_type disc|ram  # 修改集群节点的存储形式
sudo rabbitmqctl forget_cluster_node [--offline]  # 忘记节点（摘除节点）
sudo rabbitmqctl rename_cluster_node oldnode1 newnode1 [oldnode2] [newnode2...]  # 修改节点名称

# 以下 vhostpath 需替换为虚拟主机路径通过 list_vhosts 命令可进行查看

sudo rabbitmqctl status # 节点状态

sudo rabbitmqctl add_user username password # 添加用户

sudo rabbitmqctl list_users # 列出所有用户

sudo rabbitmqctl delete_user username # 删除用户

sudo rabbitmqctl clear_permissions -p vhostpath username # 清除用户权限

sudo rabbitmqctl list_user_permissions username # 列出用户权限

sudo rabbitmqctl change_password username newpassword # 修改密码

sudo rabbitmqctl set_permissions -p vhostpath username ".*" ".*" ".*" # 设置用户权限

sudo rabbitmqctl add_vhost vhostpath # 创建虚拟主机

sudo rabbitmqctl list_vhosts # 列出所有虚拟主机

sudo rabbitmqctl list_permissions -p vhostpath # 列出虚拟主机上所有权限

sudo rabbitmqctl delete_vhost vhostpath # 删除虚拟主机

sudo rabbitmqctl list_queues # 查看所有队列信息

sudo rabbitmqctl -p vhostpath purge_queue blue # 清除队列里的消息

sudo rabbitmqctl reset # 移除所有数据，要在 rabbitmqctl stop_app之后使用

sudo rabbitmqctl join_cluster <clusternode> [--ram|disc] # 组成集群命令

sudo rabbitmqctl cluster_status # 查看集群状态

sudo rabbitmqctl change_cluster_node_type disc|ram # 修改集群节点的存储形式

sudo rabbitmqctl forget_cluster_node [--offline] # 忘记节点（摘除节点）

sudo rabbitmqctl rename_cluster_node oldnode1 newnode1 [oldnode2] [newnode2...] # 修改节点名称

消息生产与消费

ConnectionFactory：获取连接工厂

Connection：获取一个连接

Channel：数据通信信道，可发送和接收消息

Queue：具体的消息存储队列

Producer & Consumer：生产和消费者

示例代码：

public class Producer {	
	public static void main(String[] args) throws Exception {		
		// 1、创建一个 ConnectionFactory并进行配置
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("192.168.xxx.xxx");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		// 2、通过连接工厂创建连接
		Connection connection = connectionFactory.newConnection();
		// 3、通过 connection 创建一个 Channel
		Channel channel = connection.createChannel();
		// 4、通过 channel 发送数据
		for(int i=0; i<5; i++) {
			String msg = "Hello RabbitMQ";
			channel.basicPublish("", "test001", null, msg.getBytes());
		}
		// 5、关闭相关连接
		channel.close();
		connection.close();
	}


#  消费者
public class Consumer {
	public static void main(String[] args) throws Exception {
		...1,2,3步代码与 Producer 相同，此处省略....
		// 4、声明一个队列
		String queueName = "test001";
		channel.queueDeclare(queueName, true, false, false, null);
		// 5、创建消费者
		QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
		// 6、设置 Channel	
		// channel.queueBind(queueName,"test_consumer_exchange","topic");
		 channel.basicConsume(queueName, true, queueingConsumer);
		 while(true) {
			 // 7、获取消息
			 Delivery delivery = queueingConsumer.nextDelivery();
			 String msg = new String(delivery.getBody());
			 System.out.println("消费值："+msg);
			 // Envelope envelope = delivery.getEnvelope();
		 }	
	}	
}

public class Producer {

public static void main(String[] args) throws Exception {

// 1、创建一个 ConnectionFactory并进行配置

ConnectionFactory connectionFactory = new ConnectionFactory();

connectionFactory.setHost("192.168.xxx.xxx");

connectionFactory.setPort(5672);

connectionFactory.setVirtualHost("/");

// 2、通过连接工厂创建连接

Connection connection = connectionFactory.newConnection();

// 3、通过 connection 创建一个 Channel

Channel channel = connection.createChannel();

// 4、通过 channel 发送数据

for(int i=0; i<5; i++) {

String msg = "Hello RabbitMQ";

channel.basicPublish("", "test001", null, msg.getBytes());

}

// 5、关闭相关连接

channel.close();

connection.close();

}

# 消费者

public class Consumer {

public static void main(String[] args) throws Exception {

...1,2,3步代码与 Producer 相同，此处省略....

// 4、声明一个队列

String queueName = "test001";

channel.queueDeclare(queueName, true, false, false, null);

// 5、创建消费者

QueueingConsumer queueingConsumer = new QueueingConsumer(channel);

// 6、设置 Channel

// channel.queueBind(queueName,"test_consumer_exchange","topic");

channel.basicConsume(queueName, true, queueingConsumer);

while(true) {

// 7、获取消息

Delivery delivery = queueingConsumer.nextDelivery();

String msg = new String(delivery.getBody());

System.out.println("消费值："+msg);

// Envelope envelope = delivery.getEnvelope();

}

注：未指定 exchange 时会使用默认AMQP default，但要求队列名称与routing key 相同（示例中为 test001）

Exchange 交换机

接收消息，并根据路由键转发消息所绑定的队列

Name：交换机名称

Type：交换机类型 direct、topic、fanout、headers

Direct：RouteKey 需匹配

# Producer
String exchangeName = "test_direct_exchange";
String routingKey = "test.direct";
String msg = "Hello World RabbitMQ 4 Direct Exchnage Message...";
channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());

# Consumer
String exchangeName ="test_direct_exchange";
String exchangeType = "direct";
String queueName = "test_direct_queue";
String routingKey = "test.direct";
channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, consumer);

# Producer

String exchangeName = "test_direct_exchange";

String routingKey = "test.direct";

String msg = "Hello World RabbitMQ 4 Direct Exchnage Message...";

channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());

# Consumer

String exchangeName ="test_direct_exchange";

String exchangeType = "direct";

String queueName = "test_direct_queue";

String routingKey = "test.direct";

channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);

channel.queueDeclare(queueName, false, false, false, null);

channel.queueBind(queueName, exchangeName, routingKey);

QueueingConsumer consumer = new QueueingConsumer(channel);

channel.basicConsume(queueName, true, consumer);

Topic：模糊匹配RouteKey，#”表示0个或若干个关键字，“*”表示一个关键字。如“log.*”能与“log.warn”匹配，无法与“log.warn.timeout”匹配；但是“log.#”能与上述两者匹配

# Producer
String exchangeName = "test_topic_exchange";
String routingKey1 = "user.save";
String routingKey2 = "user.update";
String routingKey3 = "user.delete.abc";
//发送
String msg = "Hello World RabbitMQ 4 Topic Exchnage Message...";
channel.basicPublish(exchangeName, routingKey1, null, msg.getBytes());
channel.basicPublish(exchangeName, routingKey2, null, msg.getBytes());
channel.basicPublish(exchangeName, routingKey3, null, msg.getBytes());
// 5、关闭相关连接
channel.close();
connection.close();

# Consumer
String exchangeName = "test_topic_exchange";
String exchangeType = "topic";
String queueName = "test_topic_queue";
String routingKey = "user.#";
channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, consumer);
while(true) {
	 Delivery delivery = consumer.nextDelivery();
	 String msg = new String(delivery.getBody());
	 System.out.println("收到消息："+msg);
}

# Producer

String exchangeName = "test_topic_exchange";

String routingKey1 = "user.save";

String routingKey2 = "user.update";

String routingKey3 = "user.delete.abc";

//发送

String msg = "Hello World RabbitMQ 4 Topic Exchnage Message...";

channel.basicPublish(exchangeName, routingKey1, null, msg.getBytes());

channel.basicPublish(exchangeName, routingKey2, null, msg.getBytes());

channel.basicPublish(exchangeName, routingKey3, null, msg.getBytes());

// 5、关闭相关连接

channel.close();

connection.close();

# Consumer

String exchangeName = "test_topic_exchange";

String exchangeType = "topic";

String queueName = "test_topic_queue";

String routingKey = "user.#";

channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);

channel.queueDeclare(queueName, false, false, false, null);

channel.queueBind(queueName, exchangeName, routingKey);

QueueingConsumer consumer = new QueueingConsumer(channel);

channel.basicConsume(queueName, true, consumer);

while(true) {

Delivery delivery = consumer.nextDelivery();

String msg = new String(delivery.getBody());

System.out.println("收到消息："+msg);

}

Fanout：不要处理 RouteKey，只要将简单的队列绑定到交换机上，发送到交换机的消息都会转发到与交换机绑定的队列上；不使用路由规则，因此最快

# Consumer
String exchangeName = "test_fanout_exchange";
String exchangeType = "fanout";
String queueName = "test_fanout_queue";
String routingKey = ""; // 不设置路由键
channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, consumer);
while(true) {
	 Delivery delivery = consumer.nextDelivery();
	 String msg = new String(delivery.getBody());
	 System.out.println("收到消息："+msg);
}

# Producer
String exchangeName = "test_fanout_exchange";
String routingKey = "xxxxx"; // 设置与不设置均不产生影响
//发送
String msg = "Hello World RabbitMQ 4 FANOUT Exchnage Message...";
channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());
// 5、关闭相关连接
channel.close();
connection.close();

# Consumer

String exchangeName = "test_fanout_exchange";

String exchangeType = "fanout";

String queueName = "test_fanout_queue";

String routingKey = ""; // 不设置路由键

channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);

channel.queueDeclare(queueName, false, false, false, null);

channel.queueBind(queueName, exchangeName, routingKey);

QueueingConsumer consumer = new QueueingConsumer(channel);

channel.basicConsume(queueName, true, consumer);

while(true) {

Delivery delivery = consumer.nextDelivery();

String msg = new String(delivery.getBody());

System.out.println("收到消息："+msg);

}

# Producer

String exchangeName = "test_fanout_exchange";

String routingKey = "xxxxx"; // 设置与不设置均不产生影响

//发送

String msg = "Hello World RabbitMQ 4 FANOUT Exchnage Message...";

channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());

// 5、关闭相关连接

channel.close();

connection.close();

Durability：是否需要持久化，true 为持久化

Auto delete：当最后一个绑定到 Exchange 上的队列删除后，自动删除该Exchange

Internal：当前 Exchange 是否用于 RabbitMQ 内部使用，默认为 False

Arguments：扩展参数，用于扩展 AMQP 协议自定义使用

主要概念

Binding-绑定：Exchange 和 Exchange、Queue 之间的连接关系

Queue-消息队列：实际存储消息，Durability设置是否持久化

Message-消息：服务器和应用程序之间传送的数据，由 Properties 和 Payload（Body ）组成

常用属性：delivery mode, headers（自定义属性）
其它属性：content_type, content_encoding, priority, correlation_id, reply_to, expiration, message_id, timestamp, type, user_id, app_id, cluster_id

Virtual Host-虚拟主机：逻辑隔离，最上层的消息路由，一个 VirtualHost 中不能有相同名称的 Exchange 或 Queue

属性设置及获取

# Producer
Map<String, Object> headers = new HashMap<>();
headers.put("my1", "111");
headers.put("my2", "222");
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
	.deliveryMode(2)
	.contentEncoding("UTF-8")
	.expiration("10000")
	.headers(headers)
	.build();
for(int i=0; i<5; i++) { 
	String msg = "Hello RabbitMQ"; 
	channel.basicPublish("", "test001", properties, msg.getBytes()); 
} 

# Consumer
Map<String, Object> headers =delivery.getProperties().getHeaders();
System.err.println("Header get my1 value:" + headers.get("my1"));

# Producer

Map<String, Object> headers = new HashMap<>();

headers.put("my1", "111");

headers.put("my2", "222");

AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()

.deliveryMode(2)

.contentEncoding("UTF-8")

.expiration("10000")

.headers(headers)

.build();

for(int i=0; i<5; i++) {

String msg = "Hello RabbitMQ";

channel.basicPublish("", "test001", properties, msg.getBytes());

}

# Consumer

Map<String, Object> headers =delivery.getProperties().getHeaders();

System.err.println("Header get my1 value:" + headers.get("my1"));

消息如何保障100%的投递成功？

生产端的可靠性投递

保障消息的成功发出
保障 MQ 节点的成功接收
发送端收到 MQ 节点（Broker）确认应答
完善的消息进行补偿机制

BAT/TMD 互联网大厂的解决方案

消息落库，对消息状态进行打标
消息的延迟投递，做二次确认，回调检查

幂等性

借鉴数据库的乐观锁机制讲解
比如我们执行一条更新库存的 SQL 语句：

UPDATE T_REPS SET COUNT = COUNT - 1, VERSTION = VERSION + 1 WHERE VERSION = 1

1
2
3

UPDATE T_REPS SET COUNT = COUNT - 1,
VERSTION = VERSION + 1
WHERE VERSION = 1

幂等性是指执行100次1000次…结果都相同

在海量订单产生的业务高峰期，如何避免消息的重复消费问题？

业界主流幂等性操作：

唯一 ID + 指纹码机制，利用数据库主键去重
SELECT COUNT(1) FROM T_ORDER WHERE ID = 唯一 ID + 指纹码
实现简单，但高并发下有数据库写入的性能瓶颈；可通过 ID 进行分库分表进行算法路由
利用 Redis 的原子性实现
如果进行数据落库，如何做到数据库和缓存的原子性？
若不进行数据落库，如果设置定时同步的策略？

Confirm 确认消息实现

第一步：在 channel 上开启确认模式：channel.confirmSelect()
第二步：在 channel 上添加监听：addConfirmListener

Return 消息机制

Return Listener 用于处理一些不可路由的消息(channel.addReturnListener)，channel.basicPublish中的 mandatory 参数应设为 true，默认值 false 会删除找不到路由的消息

消费端自定义监听

# 生产中不使用 queueingConsumer 而是会去自定义，Consumer 类的修改为如下
channel.basicConsume(queueName, true, new MyConsumer(channel));

# 新增MyConsumer的实现
public class MyConsumer extends DefaultConsumer {

	public MyConsumer(Channel channel) {
		super(channel);
	}

	@Override
	public void handleDelivery(String consumerTag,
        Envelope envelope,
        AMQP.BasicProperties properties,
        byte[] body) throws IOException {
	... ...	
	}
}

# 生产中不使用 queueingConsumer 而是会去自定义，Consumer 类的修改为如下

channel.basicConsume(queueName, true, new MyConsumer(channel));

# 新增MyConsumer的实现

public class MyConsumer extends DefaultConsumer {

public MyConsumer(Channel channel) {

super(channel);

}

@Override

public void handleDelivery(String consumerTag,

Envelope envelope,

AMQP.BasicProperties properties,

byte[] body) throws IOException {

... ...

}

消费端限流

RabbitMQ 提供了一种 qos（服务质量保证）功能

void BaiscQos(uint prefetchSize, ushort prefetchCount, bool global);

prefetchSize：0为不限制大小
prefetchCount：同时给消费推送多少条消息，通常设为1，在autoAck=false 的情况下才生效，即非自动签收
global：是否将设置应用于 channel 级别，false 为 consumer 级别

死信队列

DLX, Dead-Letter-Exchange，当消息在一个队列中变成死信后，它能被重新 publish 到另一个 Exchange，这个 Exchange就是 DLX

消息变成死信的几种情况：

消息被拒绝（basic.reject / basic.nack），并且requeue=false
消息 TTL 过期
队列达到最大长度

正常声明交换机、队列、绑定，只需在队列上添加参数 arguments.put(“x-dead-letter-exchange”, “dlx.exchange”);

6种模式汇总(演示代码 Golang)

https://www.rabbitmq.com/getstarted.html

Simple简单模式

最简单常用的模式

package RabbitMQ

import (
	"fmt"
	"github.com/streadway/amqp"
	"log"
)

// 格式：amqp://账号:密码@服务器地址:端口号/vhost
const MQURL = "amqp://demo:demopass@127.0.0.1:5672/demo"

type RabbitMQ struct {
	conn *amqp.Connection
	channel *amqp.Channel
	// 队列名称
	QueueName string
	// 交换机
	Exchange string
	Key string
	// 连接信息
	Mqurl string
}

// 创建 RabbitMQ 结构体实例
func NewRabbitMQ(queueName string, exhange string, key string) *RabbitMQ {
	rabbitmq := &RabbitMQ{QueueName:queueName, Exchange:exhange, Key:key, Mqurl:MQURL}
	var err error
	// 创建RabbitMQ连接
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnError(err, "创建连接错误")
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnError(err, "获取channel失败")
	return rabbitmq

}

// 断开连接
func (r *RabbitMQ) Destroy() {
	r.channel.Close()
	r.conn.Close()
}

// 错误处理函数
func (r *RabbitMQ) failOnError(err error, message string)  {
	if err != nil {
		log.Fatalf("%s:%s", message, err)
		panic(fmt.Sprintf("%s:%s", message, err))
	}
}

// 简单模式 Step 1、创建简单模式RabbitMQ 实例
func NewRabbitMQSimple(queueName string) *RabbitMQ  {
	return NewRabbitMQ(queueName, "", "")
}

// 简单模式 Step 2、简单模式生产代码
func (r *RabbitMQ) PublishSimple(message string) {
	// 1、申请队列，如果不存在自动创建，存在则跳过
	_, err := r.channel.QueueDeclare(
		r.QueueName,
		// 是否持久化
		false,
		// 是否为自动删除
		false,
		// 是否具有排他性
		false,
		// 是否阻塞
		false,
		// 额外属性
		nil)
	if err != nil {
		fmt.Println(err)
	}
	// 2、发送消息到队列中
	r.channel.Publish(
		r.Exchange,
		r.QueueName,
		// 若为true，根据Exchange类型和routingKey规则，无法找到满足条件的队列则将消息返还发送者
		false,
		// 若为true，队列未绑定消费者时将消息返还发送者
		false,
		amqp.Publishing{
			ContentType:"text/plain",
			Body:[]byte(message),
		})
}

func (r *RabbitMQ) ConsumeSimple() {
	// 1、申请队列，如果不存在自动创建，存在则跳过
	_, err := r.channel.QueueDeclare(
		r.QueueName,
		// 是否持久化
		false,
		// 是否为自动删除
		false,
		// 是否具有排他性
		false,
		// 是否阻塞
		false,
		// 额外属性
		nil)
	if err != nil {
		fmt.Println(err)
	}

	// 接收消息
	msgs, err := r.channel.Consume(
		 r.QueueName,
		 // 用于区分消息费
		 "",
		 // 是否自动应答
		 true,
		// 是否具有排他性
		false,
		// 若为true，不能将消息发给同一个连接中的消费者
		false,
		// 消费是否阻塞
		 false,
		// 额外属性
		nil)
	if err != nil {
		fmt.Println(err)
	}

	forever := make(chan bool)
	// 启用协程处理消息
	go func() {
		for d := range msgs {
			// 实现处理逻辑函数
			log.Printf("Received a message: %s", d.Body)
		}
	}()

	log.Print("[*] Waiting for messages, press Ctrl+C to exit")
	<- forever
}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

package RabbitMQ

import (

"fmt"

"github.com/streadway/amqp"

"log"

)

// 格式：amqp://账号:密码@服务器地址:端口号/vhost

const MQURL = "amqp://demo:demopass@127.0.0.1:5672/demo"

type RabbitMQ struct {

conn *amqp.Connection

channel *amqp.Channel

// 队列名称

QueueName string

// 交换机

Exchange string

Key string

// 连接信息

Mqurl string

}

// 创建 RabbitMQ 结构体实例

func NewRabbitMQ(queueName string, exhange string, key string) *RabbitMQ {

rabbitmq := &RabbitMQ{QueueName:queueName, Exchange:exhange, Key:key, Mqurl:MQURL}

var err error

// 创建RabbitMQ连接

rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)

rabbitmq.failOnError(err, "创建连接错误")

rabbitmq.channel, err = rabbitmq.conn.Channel()

rabbitmq.failOnError(err, "获取channel失败")

return rabbitmq

}

// 断开连接

func (r *RabbitMQ) Destroy() {

r.channel.Close()

r.conn.Close()

}

// 错误处理函数

func (r *RabbitMQ) failOnError(err error, message string) {

if err != nil {

log.Fatalf("%s:%s", message, err)

panic(fmt.Sprintf("%s:%s", message, err))

}

// 简单模式 Step 1、创建简单模式RabbitMQ 实例

func NewRabbitMQSimple(queueName string) *RabbitMQ {

return NewRabbitMQ(queueName, "", "")

}

// 简单模式 Step 2、简单模式生产代码

func (r *RabbitMQ) PublishSimple(message string) {

// 1、申请队列，如果不存在自动创建，存在则跳过

_, err := r.channel.QueueDeclare(

r.QueueName,

// 是否持久化

false,

// 是否为自动删除

false,

// 是否具有排他性

false,

// 是否阻塞

false,

// 额外属性

nil)

if err != nil {

fmt.Println(err)

}

// 2、发送消息到队列中

r.channel.Publish(

r.Exchange,

r.QueueName,

// 若为true，根据Exchange类型和routingKey规则，无法找到满足条件的队列则将消息返还发送者

false,

// 若为true，队列未绑定消费者时将消息返还发送者

false,

amqp.Publishing{

ContentType:"text/plain",

Body:[]byte(message),

})

}

func (r *RabbitMQ) ConsumeSimple() {

// 1、申请队列，如果不存在自动创建，存在则跳过

_, err := r.channel.QueueDeclare(

r.QueueName,

// 是否持久化

false,

// 是否为自动删除

false,

// 是否具有排他性

false,

// 是否阻塞

false,

// 额外属性

nil)

if err != nil {

fmt.Println(err)

}

// 接收消息

msgs, err := r.channel.Consume(

r.QueueName,

// 用于区分消息费

"",

// 是否自动应答

true,

// 是否具有排他性

false,

// 若为true，不能将消息发给同一个连接中的消费者

false,

// 消费是否阻塞

false,

// 额外属性

nil)

if err != nil {

fmt.Println(err)

}

forever := make(chan bool)

// 启用协程处理消息

go func() {

for d := range msgs {

// 实现处理逻辑函数

log.Printf("Received a message: %s", d.Body)

}

}()

log.Print("[*] Waiting for messages, press Ctrl+C to exit")

<- forever

}

生产者调用：

func main() {
	rabbitmq := RabbitMQ.NewRabbitMQSimple("demoSimple")
	rabbitmq.PublishSimple("Hello world!")
	fmt.Println("发送成功")
}

func main() {

rabbitmq := RabbitMQ.NewRabbitMQSimple("demoSimple")

rabbitmq.PublishSimple("Hello world!")

fmt.Println("发送成功")

}

消费者调用：

func main()  {
	rabbitmq := RabbitMQ.NewRabbitMQSimple("demoSimple")
	rabbitmq.ConsumeSimple()
}

func main() {

rabbitmq := RabbitMQ.NewRabbitMQSimple("demoSimple")

rabbitmq.ConsumeSimple()

}

Work 工作模式

一条消息只能被一个消费者获取

生产者调用：

func main() {
	rabbitmq := RabbitMQ.NewRabbitMQSimple("demoSimple")
	for i := 0; i <= 100; i++ {
		rabbitmq.PublishSimple("Hello world!")
		time.Sleep(1 * time.Second)
		fmt.Println("发送成功")
	}
}

func main() {

rabbitmq := RabbitMQ.NewRabbitMQSimple("demoSimple")

for i := 0; i <= 100; i++ {

rabbitmq.PublishSimple("Hello world!")

time.Sleep(1 * time.Second)

fmt.Println("发送成功")

}

消费者调用（起多个服务）：

func main()  {
	rabbitmq := RabbitMQ.NewRabbitMQSimple("demoSimple")
	rabbitmq.ConsumeSimple()
}

func main() {

rabbitmq := RabbitMQ.NewRabbitMQSimple("demoSimple")

rabbitmq.ConsumeSimple()

}

Publish/Subscribe 订阅模式

消息被路由投递给多个队列，一个消息被多个消费者获取。

// 订阅模式创建RabbitMQ实例
func NewRabbitMQPubSub(exhangeName string) *RabbitMQ {
	rabbitmq := NewRabbitMQ("", exhangeName, "")
	var err error
	// 获取 connection
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnError(err, "创建连接错误")
	// 获取channel
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnError(err, "获取channel失败")
	return rabbitmq
}

// 订阅模式生产
func (r *RabbitMQ) PublishPub(message string) {
	// 尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		"fanout",
		true,
		false,
		// true表示这个exchange不能被client用来推送消息，仅用于exchange之间的绑定
		false,
		false,
		nil)
	r.failOnError(err, "Failed to declare an exchange")
	// 发送消息
	err = r.channel.Publish(
		r.Exchange,
		"",
		false,
		false,
		amqp.Publishing{
			ContentType:"text/plain",
			Body:[]byte(message),
		})
}

// 订阅模式消费端代码
func (r *RabbitMQ) ReceiveSub() {
	// 尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		// 交换机类型
		"fanout",
		true,
		false,
		false,
		false,
		nil)
	r.failOnError(err, "Failed to declare an exchange")

	// 尝试创建队列，这里队列名称不要写
	q, err := r.channel.QueueDeclare(
		"", // 随机生产队列名称
		false,
		false,
		true,
		false,
		nil)
	r.failOnError(err, "Failed to declare a queue")

	// 绑定队列到exchange中
	err = r.channel.QueueBind(
		q.Name,
		// pub/sub模式下，key应为空
		"",
		r.Exchange,
		 false,
		 nil)

	// 消费消息
	messages, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil)

	forever := make(chan bool)

	go func() {
		for d := range messages {
			log.Printf("Received a message: %s", d.Body)
		}
	}()
	log.Print("[*] Waiting for messages, press Ctrl+C to exit")
	<- forever
}

// 订阅模式创建RabbitMQ实例

func NewRabbitMQPubSub(exhangeName string) *RabbitMQ {

rabbitmq := NewRabbitMQ("", exhangeName, "")

var err error

// 获取 connection

rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)

rabbitmq.failOnError(err, "创建连接错误")

// 获取channel

rabbitmq.channel, err = rabbitmq.conn.Channel()

rabbitmq.failOnError(err, "获取channel失败")

return rabbitmq

}

// 订阅模式生产

func (r *RabbitMQ) PublishPub(message string) {

// 尝试创建交换机

err := r.channel.ExchangeDeclare(

r.Exchange,

"fanout",

true,

false,

// true表示这个exchange不能被client用来推送消息，仅用于exchange之间的绑定

false,

nil)

r.failOnError(err, "Failed to declare an exchange")

// 发送消息

err = r.channel.Publish(

r.Exchange,

"",

false,

amqp.Publishing{

ContentType:"text/plain",

Body:[]byte(message),

})

}

// 订阅模式消费端代码

func (r *RabbitMQ) ReceiveSub() {

// 尝试创建交换机

err := r.channel.ExchangeDeclare(

r.Exchange,

// 交换机类型

"fanout",

true,

false,

nil)

r.failOnError(err, "Failed to declare an exchange")

// 尝试创建队列，这里队列名称不要写

q, err := r.channel.QueueDeclare(

"", // 随机生产队列名称

false,

true,

false,

nil)

r.failOnError(err, "Failed to declare a queue")

// 绑定队列到exchange中

err = r.channel.QueueBind(

q.Name,

// pub/sub模式下，key应为空

"",

r.Exchange,

false,

nil)

// 消费消息

messages, err := r.channel.Consume(

q.Name,

"",

true,

false,

nil)

forever := make(chan bool)

go func() {

for d := range messages {

log.Printf("Received a message: %s", d.Body)

}

}()

log.Print("[*] Waiting for messages, press Ctrl+C to exit")

<- forever

}

生产者调用：

func main() {
   rabbitmq := RabbitMQ.NewRabbitMQPubSub("newProduct")
   for i := 0; i< 100; i++ {
      rabbitmq.PublishPub("订阅模式生产第" + strconv.Itoa(i) + "条数据")
      fmt.Println("订阅模式生产第" + strconv.Itoa(i) + "条数据")
      time.Sleep(1 * time.Second)
   }
}

func main() {

rabbitmq := RabbitMQ.NewRabbitMQPubSub("newProduct")

for i := 0; i< 100; i++ {

rabbitmq.PublishPub("订阅模式生产第" + strconv.Itoa(i) + "条数据")

fmt.Println("订阅模式生产第" + strconv.Itoa(i) + "条数据")

time.Sleep(1 * time.Second)

}

消费者调用（起多个服务）：

func main()  {
   rabbitmq := RabbitMQ.NewRabbitMQPubSub("newProduct")
   rabbitmq.ReceiveSub()
}

func main() {

rabbitmq := RabbitMQ.NewRabbitMQPubSub("newProduct")

rabbitmq.ReceiveSub()

}

Routing路由模式

一个消息被多个消费者获取。并且消息的目标队列可被生产者指定。

// 路由模式创建RabbitMQ实例
func NewRabbitMQRouting(exhangeName string, routingKey string) *RabbitMQ {
	rabbitmq := NewRabbitMQ("", exhangeName, routingKey)
	var err error
	// 获取 connection
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnError(err, "创建连接错误")
	// 获取channel
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnError(err, "获取channel失败")
	return rabbitmq
}

// 路由模式生产
func (r *RabbitMQ) PublishRouting(message string) {
	// 尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		// 修改为direct
		"direct",
		true,
		false,
		// true表示这个exchange不能被client用来推送消息，仅用于exchange之间的绑定
		false,
		false,
		nil)
	r.failOnError(err, "Failed to declare an exchange")
	// 发送消息
	err = r.channel.Publish(
		r.Exchange,
		r.Key,
		false,
		false,
		amqp.Publishing{
			ContentType:"text/plain",
			Body:[]byte(message),
		})
}

// 路由模式消费端代码
func (r *RabbitMQ) ReceiveRouting() {
	// 尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		// 交换机类型
		"direct",
		true,
		false,
		false,
		false,
		nil)
	r.failOnError(err, "Failed to declare an exchange")

	// 尝试创建队列，这里队列名称不要写
	q, err := r.channel.QueueDeclare(
		"", // 随机生产队列名称
		false,
		false,
		true,
		false,
		nil)
	r.failOnError(err, "Failed to declare a queue")

	// 绑定队列到exchange中
	err = r.channel.QueueBind(
		q.Name,
		// 需要绑定key
		r.Key,
		r.Exchange,
		false,
		nil)

	// 消费消息
	messages, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil)

	forever := make(chan bool)

	go func() {
		for d := range messages {
			log.Printf("Received a message: %s", d.Body)
		}
	}()
	log.Print("[*] Waiting for messages, press Ctrl+C to exit")
	<- forever
}

// 路由模式创建RabbitMQ实例

func NewRabbitMQRouting(exhangeName string, routingKey string) *RabbitMQ {

rabbitmq := NewRabbitMQ("", exhangeName, routingKey)

var err error

// 获取 connection

rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)

rabbitmq.failOnError(err, "创建连接错误")

// 获取channel

rabbitmq.channel, err = rabbitmq.conn.Channel()

rabbitmq.failOnError(err, "获取channel失败")

return rabbitmq

}

// 路由模式生产

func (r *RabbitMQ) PublishRouting(message string) {

// 尝试创建交换机

err := r.channel.ExchangeDeclare(

r.Exchange,

// 修改为direct

"direct",

true,

false,

// true表示这个exchange不能被client用来推送消息，仅用于exchange之间的绑定

false,

nil)

r.failOnError(err, "Failed to declare an exchange")

// 发送消息

err = r.channel.Publish(

r.Exchange,

r.Key,

false,

amqp.Publishing{

ContentType:"text/plain",

Body:[]byte(message),

})

}

// 路由模式消费端代码

func (r *RabbitMQ) ReceiveRouting() {

// 尝试创建交换机

err := r.channel.ExchangeDeclare(

r.Exchange,

// 交换机类型

"direct",

true,

false,

nil)

r.failOnError(err, "Failed to declare an exchange")

// 尝试创建队列，这里队列名称不要写

q, err := r.channel.QueueDeclare(

"", // 随机生产队列名称

false,

true,

false,

nil)

r.failOnError(err, "Failed to declare a queue")

// 绑定队列到exchange中

err = r.channel.QueueBind(

q.Name,

// 需要绑定key

r.Key,

r.Exchange,

false,

nil)

// 消费消息

messages, err := r.channel.Consume(

q.Name,

"",

true,

false,

nil)

forever := make(chan bool)

go func() {

for d := range messages {

log.Printf("Received a message: %s", d.Body)

}

}()

log.Print("[*] Waiting for messages, press Ctrl+C to exit")

<- forever

}

生产者调用：

func main() {
   demoOne := RabbitMQ.NewRabbitMQRouting("demoEx", "demo_one")
   demoTwo := RabbitMQ.NewRabbitMQRouting("demoEx", "demo_two")
   for i:=0; i<=100; i++ {
      demoOne.PublishRouting("Hello demo one!" + strconv.Itoa(i))
      demoTwo.PublishRouting("Hello demo two!" + strconv.Itoa(i))
      time.Sleep(1 * time.Second)
      fmt.Println(i)
   }
}

func main() {

demoOne := RabbitMQ.NewRabbitMQRouting("demoEx", "demo_one")

demoTwo := RabbitMQ.NewRabbitMQRouting("demoEx", "demo_two")

for i:=0; i<=100; i++ {

demoOne.PublishRouting("Hello demo one!" + strconv.Itoa(i))

demoTwo.PublishRouting("Hello demo two!" + strconv.Itoa(i))

time.Sleep(1 * time.Second)

fmt.Println(i)

}

消费者调用（起多个服务）：

// receiveRoutingOne.go
func main() {
   demoOne := RabbitMQ.NewRabbitMQRouting("demoEx", "demo_one")
   demoOne.ReceiveRouting()
}

// receiveRoutingOne.go

func main() {

demoOne := RabbitMQ.NewRabbitMQRouting("demoEx", "demo_one")

demoOne.ReceiveRouting()

}

// receiveRoutingTwo.go
func main() {
   demoTwo := RabbitMQ.NewRabbitMQRouting("demoEx", "demo_two")
   demoTwo.ReceiveRouting()
}

// receiveRoutingTwo.go

func main() {

demoTwo := RabbitMQ.NewRabbitMQRouting("demoEx", "demo_two")

demoTwo.ReceiveRouting()

}

Topic话题模式

一个消息被多个消费者获取。消息的目标队列可用BindingKey以通配符(#：一个或多个词，*：一个词)的方式指定。

// 话题模式创建RabbitMQ实例
func NewRabbitMQTopic(exhangeName string, routingKey string) *RabbitMQ {
	rabbitmq := NewRabbitMQ("", exhangeName, routingKey)
	var err error
	// 获取 connection
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnError(err, "创建连接错误")
	// 获取channel
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnError(err, "获取channel失败")
	return rabbitmq
}

// 话题模式生产
func (r *RabbitMQ) PublishTopic(message string) {
	// 尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		// 修改为topic
		"topic",
		true,
		false,
		// true表示这个exchange不能被client用来推送消息，仅用于exchange之间的绑定
		false,
		false,
		nil)
	r.failOnError(err, "Failed to declare an exchange")
	// 发送消息
	err = r.channel.Publish(
		r.Exchange,
		r.Key,
		false,
		false,
		amqp.Publishing{
			ContentType:"text/plain",
			Body:[]byte(message),
		})
}

// 话题模式消费端代码
func (r *RabbitMQ) ReceiveTopic() {
	// 尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		// 交换机类型
		"topic",
		true,
		false,
		false,
		false,
		nil)
	r.failOnError(err, "Failed to declare an exchange")

	// 尝试创建队列，这里队列名称不要写
	q, err := r.channel.QueueDeclare(
		"", // 随机生产队列名称
		false,
		false,
		true,
		false,
		nil)
	r.failOnError(err, "Failed to declare a queue")

	// 绑定队列到exchange中
	err = r.channel.QueueBind(
		q.Name,
		// 需要绑定key
		r.Key,
		r.Exchange,
		false,
		nil)

	// 消费消息
	messages, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil)

	forever := make(chan bool)

	go func() {
		for d := range messages {
			log.Printf("Received a message: %s", d.Body)
		}
	}()
	log.Print("[*] Waiting for messages, press Ctrl+C to exit")
	<- forever
}

// 话题模式创建RabbitMQ实例

func NewRabbitMQTopic(exhangeName string, routingKey string) *RabbitMQ {

rabbitmq := NewRabbitMQ("", exhangeName, routingKey)

var err error

// 获取 connection

rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)

rabbitmq.failOnError(err, "创建连接错误")

// 获取channel

rabbitmq.channel, err = rabbitmq.conn.Channel()

rabbitmq.failOnError(err, "获取channel失败")

return rabbitmq

}

// 话题模式生产

func (r *RabbitMQ) PublishTopic(message string) {

// 尝试创建交换机

err := r.channel.ExchangeDeclare(

r.Exchange,

// 修改为topic

"topic",

true,

false,

// true表示这个exchange不能被client用来推送消息，仅用于exchange之间的绑定

false,

nil)

r.failOnError(err, "Failed to declare an exchange")

// 发送消息

err = r.channel.Publish(

r.Exchange,

r.Key,

false,

amqp.Publishing{

ContentType:"text/plain",

Body:[]byte(message),

})

}

// 话题模式消费端代码

func (r *RabbitMQ) ReceiveTopic() {

// 尝试创建交换机

err := r.channel.ExchangeDeclare(

r.Exchange,

// 交换机类型

"topic",

true,

false,

nil)

r.failOnError(err, "Failed to declare an exchange")

// 尝试创建队列，这里队列名称不要写

q, err := r.channel.QueueDeclare(

"", // 随机生产队列名称

false,

true,

false,

nil)

r.failOnError(err, "Failed to declare a queue")

// 绑定队列到exchange中

err = r.channel.QueueBind(

q.Name,

// 需要绑定key

r.Key,

r.Exchange,

false,

nil)

// 消费消息

messages, err := r.channel.Consume(

q.Name,

"",

true,

false,

nil)

forever := make(chan bool)

go func() {

for d := range messages {

log.Printf("Received a message: %s", d.Body)

}

}()

log.Print("[*] Waiting for messages, press Ctrl+C to exit")

<- forever

}

生产者调用：

func main() {
   demoOne := RabbitMQ.NewRabbitMQTopic("exDemoTopic", "demo.topic.one")
   demoTwo := RabbitMQ.NewRabbitMQTopic("exDemoTopic", "demo.topic.two")
   for i := 0; i <= 10; i++ {
      demoOne.PublishTopic("Hello demo topic one!" + strconv.Itoa(i))
      demoTwo.PublishTopic("Hello demo topic two!" + strconv.Itoa(i))
      time.Sleep(1 * time.Second)
      fmt.Println(i)
   }
}

func main() {

demoOne := RabbitMQ.NewRabbitMQTopic("exDemoTopic", "demo.topic.one")

demoTwo := RabbitMQ.NewRabbitMQTopic("exDemoTopic", "demo.topic.two")

for i := 0; i <= 10; i++ {

demoOne.PublishTopic("Hello demo topic one!" + strconv.Itoa(i))

demoTwo.PublishTopic("Hello demo topic two!" + strconv.Itoa(i))

time.Sleep(1 * time.Second)

fmt.Println(i)

}

消费者调用（起多个服务）：

// receiveDemoTopicAll.go
func main() {
   demoOne := RabbitMQ.NewRabbitMQTopic("exDemoTopic", "#")
   demoOne.ReceiveTopic()
}

// receiveDemoTopicAll.go

func main() {

demoOne := RabbitMQ.NewRabbitMQTopic("exDemoTopic", "#")

demoOne.ReceiveTopic()

}

// receiveDemoTopicTwo.go
func main() {
   demoTwo := RabbitMQ.NewRabbitMQTopic("exDemoTopic", "demo.*.two")
   demoTwo.ReceiveTopic()
}

// receiveDemoTopicTwo.go

func main() {

demoTwo := RabbitMQ.NewRabbitMQTopic("exDemoTopic", "demo.*.two")

demoTwo.ReceiveTopic()

}

RPC模式

常见问题

1、ERROR: epmd error for host 10: badarg (unknown POSIX error)

# sudo vi /etc/rabbitmq/rabbitmq-env.conf
NODENAME=rabbit@localhost

sudo vi /etc/hostname # 修改主机名并重启

# sudo vi /etc/rabbitmq/rabbitmq-env.conf

NODENAME=rabbit@localhost

sudo vi /etc/hostname # 修改主机名并重启

2、{error,{not_a_dets_file,”/var/lib/rabbitmq/mnesia/rabbit@manamana/recovery.dets”}}}

删除/var/lib/rabbitmq/mnesia/rabbit@manamana/recovery.dets文件再重新启动即可