RabbitMQ 第二课 快速入门
cj0 2020-06-06
急速入门 - 消息生成与消费
1. ConnectionFactory ： 获取连接工厂
2. Connection ： 一个连接
3. Channel：数据通信信道，可发送和接收消息
4. Queue：具体的消息存储队列
5. Producer & Consumer 生产者和消费者
介绍：
RabbitMQ是开源的消息中间件，它是轻量级的，支持多种消息传递协议，可以部署在分布式和联合配置中，以满足高级别、高可用性需求。并且可在许多操作系统和云环境上运行，并为大多数流行语言提供了广泛的开发工具。（这里只介绍JAVA下的RabbitMQ的使用，感兴趣的可以查看官方文档：http://www.rabbitmq.com/getstarted.html）；
安装：
参考：http://www.cnblogs.com/lfalex0831/p/8951955.html（windows安装）
应用场景：
1、异步处理，主要为了较少请求的响应时间和解耦。即将比较耗时又不需要同步的操作放入消息队列中进行传递请求，只要保证消息格式（可以理解为接头的暗号）不变，这样消息的发送方和接收方互不干扰交互，即为解耦；
2、广播，顾名思义，广播的好处就是一次发送，大家共享，大大的减少了冗余的操作，也降低了新增消费者的成本；
3、流量削峰，比如秒杀活动，因为流量过大，导致应用挂掉,为了避免这个问题，会在应用前端加入消息队列。
作用:
1.可以控制进入后台的服务，超过阀值的订单直接丢弃；
2.可以缓解瞬时的高流量压垮应用；
ps：秒杀系统优化思路可以从将请求尽量拦截在系统上游和充分利用缓存；
（如果还有别的应用场景，请大家多多指教。。。）
使用场景（本文使用的RabbitMQ的版本为5.20版本）：
This tutorial assumes RabbitMQ is installed and running on localhost on standard port (5672). In case you use a different host, port or credentials, connections settings would require adjusting.
注意：根据官方文档说明，RabbitMQ的默认访问端口为5672，而管理端口为15672，希望不要搞混了（我刚接触时就没注意，果断乱了。。-_-||）。
基本概念：
- Broker：简单来说就是消息队列服务器实体
- Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列
- Queue：消息队列载体，每个消息都会被投入到一个或多个队列
- Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来
- Routing Key：路由关键字，exchange根据这个关键字进行消息投递
- vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离
- producer：消息生产者，就是投递消息的程序
- consumer：消息消费者，就是接受消息的程序
- channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务
1、简单使用（HelloWorld）
在下图中，P是我们的生产者，C是我们的消费者。中间的框是一个队列——RabbitMQ代表消费者保存的消息缓冲区。
创建RabbitMQ的工厂类：
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ConnectionUtil {
private static final String RABBIT_HOST = "localhost";
private static final String RABBIT_USERNAME = "xiaochao";
private static final String RABBIT_PASSWORD = "root";
private static Connection connection = null;
public static Connection getConnection() {
if(connection == null) {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(RABBIT_HOST);
connectionFactory.setUsername(RABBIT_USERNAME);
connectionFactory.setPassword(RABBIT_PASSWORD);
try {
connection = connectionFactory.newConnection();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
return connection;
}
}创建生产者Producer：
package com.everjiankang.dependency.demo2;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Producer {
private static final String QUEUE_NAME="test_queue";
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
//获取连接
Connection connection = ConnectionUtil.getConnection();
System.out.println(connection);
//创建通道
Channel channel = connection.createChannel(1);
/*
* 声明（创建）队列
* 参数1：队列名称
* 参数2：为true时server重启队列不会消失
* 参数3：队列是否是独占的，如果为true只能被一个connection使用，其他连接建立时会抛出异常
* 参数4：队列不再使用时是否自动删除（没有连接，并且没有未处理的消息)
* 参数5：建立队列时的其他参数
*/
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
String message = "Hello World!";
for (int i = 0; i < 20; i++) {
message = message + i;
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
Thread.sleep(1000);
}
System.out.println("生产者 send ："+message);
channel.close();
connection.close();
}
}创建消费者Consumer：
import com.cn.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;
public class Consumer {
private static final String QUEUE_NAME = "test_queue";
public static void main(String[] args) throws IOException {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel(1);
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
StringBuffer message = new StringBuffer();
//自4.0+ 版本后无法再使用QueueingConsumer，而官方推荐使用DefaultConsumer
com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
super.handleDelivery(consumerTag, envelope, properties, body);
message.append(new String(body,"UTF-8"));
System.out.println(new String(body,"UTF-8"));
}
};
//监听队列，当b为true时，为自动提交（只要消息从队列中获取，无论消费者获取到消息后是否成功消息，都认为是消息已经成功消费），
// 当b为false时，为手动提交（消费者从队列中获取消息后，服务器会将该消息标记为不可用状态，等待消费者的反馈，
// 如果消费者一直没有反馈，那么该消息将一直处于不可用状态。
//如果选用自动确认,在消费者拿走消息执行过程中出现宕机时,消息可能就会丢失！！）
//使用channel.basicAck(envelope.getDeliveryTag(),false);进行消息确认
channel.basicConsume(QUEUE_NAME,true,consumer);
System.out.println(message.toString());
}
}测试结果，Consumer收到Producer的消息。
2、Work Queue
上一个例子是一对一发送接收形式，而工作队列为一对多发送接收形式。工作队列(即任务队列)背后的主要思想是避免立即执行资源密集型任务，并且必须等待它完成。相反，我们把任务安排在以后做。我们将任务封装为消息并将其发送到队列。在后台运行的工作进程会弹出任务并最终执行任务。当你运行许多Consumer时，任务将在他们之间共享，如下图：
由于工厂类已经创建，直接使用即可。
创建生产者Producer：
创建消费者1，2：
测试结果，当消费者中的channel.basicQos(1);这行代码的注释打开时,执行会发现,休眠时间短的消费者执行的任务多，而注释后，再次执行会发现消费者1和消费者2获取到的消息内容是不同的，同一个消息只能被一个消费者获取，消费者1和消费者2获取到的消息的数量是相同的，一个是奇数一个是偶数。
3、Publish/Subscribe（严格来说下面介绍的路由和通配符模式也是发布订阅）
在发布订阅模式中，消息需要发送到MQ的交换机exchange上，exchange根据配置的路由方式发到相应的Queue上，Queue又将消息发送给consumer，消息从queue到consumer， 消息队列的使用过程大概如下：
1.客户端连接到消息队列服务器，打开一个channel。
2.客户端声明一个exchange，并设置相关属性。
3.客户端声明一个queue，并设置相关属性。
4.客户端在exchange和queue之间建立好绑定关系。
5.客户端投递消息到exchange。
创建生产者Producer：
创建消费者Consumer1、2：
4、Routing（路由）
根据指定的路由键发送到对应的消息队列中，如下图，在这个设置中，我们可以看到与它绑定的两个队列的直接交换X。第一个队列绑定了绑定键橙色，第二个队列有两个绑定，一个绑定键为黑色，另一个为绿色。在这样的设置中，将发送到与路由键橙色的交换的消息将被路由到队列Q1。带有黑色或绿色的路由键的消息将会进入Q2。所有其他消息将被丢弃。
创建生产者Producer：
创建消费者Consumer1、2：
5、Topics（主题通配符）
可以理解为Routing的通配符模式，如下图：
“#”：表示匹配一个或多个词；（lazy.a.b.c）
“*”：表示匹配一个词；（a.orange.b）
创建生产者Producer：
创建消费者Consumer1、2：
6、RPC（远程调用）
如果我们需要在远程计算机上运行一个函数并等待结果，这种模式通常称为远程过程调用或RPC；
创建RPC服务：
创建RPC客户端：
创建RPC测试类：
