RabbitMQ消息中间件

1.MQ

MQ全称 Message Queue（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信。

之前服务与服务之间如何通信?

Openfeign 服务与服务之间直接调用

使用MQ完成系统与系统之间得调用

2.MQ的优缺点

优点：

1. 应用解耦

2.异步提速

3.削峰填谷

缺点：

什么情况下要使用MQ

1.生产者无需从消费者获取反馈。引入消息队列前的直接调用，其接口返回值为空，这会让下层的动作还没做，上层却继续向后运行，即异步成为可能。

2.容许短暂的不一致性

3.使用后有效果。解耦、提速和削峰的收益超过它的成本。

3.MQ的种类

rabbitMQ

kafka

RocketMQActiveMQ

4.rabbitMQ安装

请参考rabbirMQ安装说明文档

注意：在linux中需要放行下面的端口号或者关闭linux的防火墙

5.rabbitMQ的端口号解释

6. rabbit的工作原理

7.java程序连接RabbitMQ服务

rabbitMQ的官网

RabbitMQ Tutorials — RabbitMQ

官网提供了五种模式

创建maven工程，引入依赖

       <dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.14.2</version></dependency>

提供了5种模式。

1. 简单模式--Hello

简单模式生产者代码：

public class Test01 {public static void main(String[] args) throws Exception{ConnectionFactory factory=new ConnectionFactory();//设置rabbitMQ服务器的地址 默认localhostfactory.setHost("192.168.226.234");// 设置rabbitMQ的端口号 默认5672factory.setPort(5672);//设置账号和密码 默认guestfactory.setUsername("guest");factory.setPassword("guest");//设置虚拟主机名 默认为 /factory.setVirtualHost("/");// 获取连接通道Connection connection=factory.newConnection();// 获取channel信道Channel channel = connection.createChannel();// 创建队列/*** 如果该队列名不存在则自动创建，存在则不创建* String queue,队列名* boolean durable,是否持久化* boolean exclusive,（独占）声明队列同一时间只能保 证一个连接，且该队列只有被这一个连接使用。* boolean autoDelete,是否自动删除* Map<String, Object> arguments: 其他参数*/channel.queueDeclare("simple_queue",true,false,false, null);//发送消息到队列/*** String exchange,把消息发给哪个交换机--简单模式没 有交换机""* String routingKey,消息绑定的路由key 如果为简单模 式 默认写为队列名称* BasicProperties props, 消息的属性* byte[] body: 消息的内容*/String msg="hello world~~~~1";channel.basicPublish("","simple_queue",null,msg.getBytes());connection.close();}
}

运行示例：

注:若出现连接超时，检查是否未放行5672端口号

简单模式消费者代码:

public class Test01 {public static void main(String[] args) throws Exception{ConnectionFactory factory = new ConnectionFactory();//设置rabbitMQ服务器的地址factory.setHost("192.168.226.234");//端口号factory.setPort(5672);//设置账号密码factory.setUsername("guest");factory.setPassword("guest");//设置虚拟主机名factory.setVirtualHost("/");//获取连接通道Connection connection = factory.newConnection();//获取channel信道Channel channel = connection.createChannel();//监听队列/*** String queue,监听的队列名称* autoAck:是否自动确认消息* Consumer callback: 监听到消息后触发的回调函数*/DefaultConsumer consumer = new DefaultConsumer(channel){//有消息就会触发该方法// body:消息内容@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接受的消息:"+new String(body));}};channel.basicConsume("simple_queue",true,consumer);}
}

2. 工作者模式--work queues

P:生产者

C1:消费者1

C2:消费者2

Q: 队列

消费者1和消费者2属于竞争关系，一个消息只会被一个消费者消费。

工作者模式的生产者代码（和简单模式没什么区别）：

public class WorkTest {public static void main(String[] args) throws Exception{ConnectionFactory factory=new ConnectionFactory();//设置rabbitMQ服务器的地址 默认localhostfactory.setHost("192.168.226.234");// 设置rabbitMQ的端口号 默认5672factory.setPort(5672);//设置账号和密码 默认guestfactory.setUsername("guest");factory.setPassword("guest");//设置虚拟主机名 默认为 /factory.setVirtualHost("/");// 获取连接通道Connection connection=factory.newConnection();// 获取channel信道Channel channel = connection.createChannel();// 创建队列/*** 如果该队列名不存在则自动创建，存在则不创建* String queue,队列名* boolean durable,是否持久化* boolean exclusive,（独占）声明队列同一时间只能保 证一个连接，且该队列只有被这一个连接使用。* boolean autoDelete,是否自动删除* Map<String, Object> arguments: 其他参数*/channel.queueDeclare("work_queue",true,false,false, null);//发送消息到队列/*** String exchange,把消息发给哪个交换机--简单模式没 有交换机""* String routingKey,消息绑定的路由key 如果为简单模 式 默认写为队列名称* BasicProperties props, 消息的属性* byte[] body: 消息的内容*/for(int i=0;i<10;i++) {String msg = "hello work~~~~"+i;channel.basicPublish("", "work_queue", null, msg.getBytes());}connection.close();}
}

工作者模式消费者代码:

消费1：

public class WorkTest01 {public static void main(String[] args) throws Exception{ConnectionFactory factory = new ConnectionFactory();//设置rabbitMQ服务器的地址factory.setHost("192.168.226.234");//端口号factory.setPort(5672);//设置账号密码factory.setUsername("guest");factory.setPassword("guest");//设置虚拟主机名factory.setVirtualHost("/");//获取连接通道Connection connection = factory.newConnection();//获取channel信道Channel channel = connection.createChannel();//监听队列/*** String queue,监听的队列名称* autoAck:是否自动确认消息* Consumer callback: 监听到消息后触发的回调函数*/DefaultConsumer consumer = new DefaultConsumer(channel){//有消息就会触发该方法// body:消息内容@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接受的消息:"+new String(body));}};channel.basicConsume("work_queue",true,consumer);}
}

消费者2：

public class WorkTest02 {public static void main(String[] args) throws Exception{ConnectionFactory factory = new ConnectionFactory();//设置rabbitMQ服务器的地址factory.setHost("192.168.226.234");//端口号factory.setPort(5672);//设置账号密码factory.setUsername("guest");factory.setPassword("guest");//设置虚拟主机名factory.setVirtualHost("/");//获取连接通道Connection connection = factory.newConnection();//获取channel信道Channel channel = connection.createChannel();//监听队列/*** String queue,监听的队列名称* autoAck:是否自动确认消息* Consumer callback: 监听到消息后触发的回调函数*/DefaultConsumer consumer = new DefaultConsumer(channel){//有消息就会触发该方法// body:消息内容@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接受的消息:"+new String(body));}};channel.basicConsume("work_queue",true,consumer);}
}

两个消费者采用轮询策略

3. 发布订阅模式

p: producter 生产者

x：exchange交换机

Q: 队列

C1和C2:消费者

生产者：

//发布订阅模式
public class TestSubscribe {public static void main(String[] args) throws Exception{ConnectionFactory factory=new ConnectionFactory();//设置rabbitMQ服务器的地址 默认localhostfactory.setHost("192.168.226.234");// 设置rabbitMQ的端口号 默认5672factory.setPort(5672);//设置账号和密码 默认guestfactory.setUsername("guest");factory.setPassword("guest");//设置虚拟主机名 默认为 /factory.setVirtualHost("/");// 获取连接通道Connection connection=factory.newConnection();// 获取channel信道Channel channel = connection.createChannel();// 创建队列/*** 如果该队列名不存在则自动创建，存在则不创建* String queue,队列名* boolean durable,是否持久化* boolean exclusive,（独占）声明队列同一时间只能保 证一个连接，且该队列只有被这一个连接使用。* boolean autoDelete,是否自动删除* Map<String, Object> arguments: 其他参数*/channel.queueDeclare("publisher_queue01",true,false,false, null);channel.queueDeclare("publisher_queue01",true,false,false, null);//创建交换机/*** String exchange,交换机的名称* BuiltinExchangeType type,交换机的种类* boolean durable:是否持久化*/channel.exchangeDeclare("fanout_exchange", BuiltinExchangeType.FANOUT,true);//交换机和队列绑定/*** String queue,队列名* String exchange,交换机名* String routingKey 路由key*/channel.queueBind("publisher_queue01","fanout_exchange","");channel.queueBind("publisher_queue02","fanout_exchange","");//发送消息到队列/*** String exchange,把消息发给哪个交换机--简单模式没 有交换机""* String routingKey,消息绑定的路由key 如果为简单模 式 默认写为队列名称* BasicProperties props, 消息的属性* byte[] body: 消息的内容*/for(int i=0;i<10;i++) {String msg = "hello work~~~~"+i;channel.basicPublish("", "fanout_exchange", null, msg.getBytes());}connection.close();}
}

消费者01：

public class TestSubscribe {public static void main(String[] args) throws Exception{ConnectionFactory factory = new ConnectionFactory();//设置rabbitMQ服务器的地址factory.setHost("192.168.226.234");//端口号factory.setPort(5672);//设置账号密码factory.setUsername("guest");factory.setPassword("guest");//设置虚拟主机名factory.setVirtualHost("/");//获取连接通道Connection connection = factory.newConnection();//获取channel信道Channel channel = connection.createChannel();//监听队列/*** String queue,监听的队列名称* autoAck:是否自动确认消息* Consumer callback: 监听到消息后触发的回调函数*/DefaultConsumer consumer = new DefaultConsumer(channel){//有消息就会触发该方法// body:消息内容@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接受的消息:"+new String(body));}};channel.basicConsume("publisher_queue01",true,consumer);}
}

4. 路由模式--router

p:生产者

x: 交换机---Direct (路由模式)

c1和c2表示消费者：

Q:队列

生产者;

public class Test {public static void main(String[] args) throws Exception{ConnectionFactory factory=new ConnectionFactory();//设置rabbitMQ服务器的地址 默认localhostfactory.setHost("192.168.226.234");// 设置rabbitMQ的端口号 默认5672factory.setPort(5672);//设置账号和密码 默认guestfactory.setUsername("guest");factory.setPassword("guest");//设置虚拟主机名 默认为 /factory.setVirtualHost("/");// 获取连接通道Connection connection=factory.newConnection();// 获取channel信道Channel channel = connection.createChannel();// 创建队列/*** 如果该队列名不存在则自动创建，存在则不创建* String queue,队列名* boolean durable,是否持久化* boolean exclusive,（独占）声明队列同一时间只能保 证一个连接，且该队列只有被这一个连接使用。* boolean autoDelete,是否自动删除* Map<String, Object> arguments: 其他参数*/channel.queueDeclare("router_queue01",true,false,false, null);channel.queueDeclare("router_queue01",true,false,false, null);//创建交换机/*** String exchange,交换机的名称* BuiltinExchangeType type,交换机的种类* boolean durable:是否持久化*/channel.exchangeDeclare("direct_exchange", BuiltinExchangeType.DIRECT,true);//交换机和队列绑定/*** String queue,队列名* String exchange,交换机名* String routingKey 路由key*/channel.queueBind("router_queue01","direct_exchange","error");channel.queueBind("router_queue02","direct_exchange","error");channel.queueBind("router_queue02","direct_exchange","info");channel.queueBind("router_queue02","direct_exchange","warning");//发送消息到队列/*** String exchange,把消息发给哪个交换机--简单模式没 有交换机""* String routingKey,消息绑定的路由key 如果为简单模 式 默认写为队列名称* BasicProperties props, 消息的属性* byte[] body: 消息的内容*/String msg = "hello router~~~~";channel.basicPublish("direct_exchange", "info", null, msg.getBytes());connection.close();}
}

消费者;

public class Test {public static void main(String[] args) throws Exception{ConnectionFactory factory = new ConnectionFactory();//设置rabbitMQ服务器的地址factory.setHost("192.168.226.234");//端口号factory.setPort(5672);//设置账号密码factory.setUsername("guest");factory.setPassword("guest");//设置虚拟主机名factory.setVirtualHost("/");//获取连接通道Connection connection = factory.newConnection();//获取channel信道Channel channel = connection.createChannel();//监听队列/*** String queue,监听的队列名称* autoAck:是否自动确认消息* Consumer callback: 监听到消息后触发的回调函数*/DefaultConsumer consumer = new DefaultConsumer(channel){//有消息就会触发该方法// body:消息内容@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接受的消息:"+new String(body));}};channel.basicConsume("router_queue01",true,consumer);}
}

5. 主题模式--topic

* 通配一个单词

# 通配零个或多个单词