一 基本介绍
1.1 可以解决什么问题
- 异步处理
- 应用解耦
- 流量削峰
- 日志处理
1.2 安装
学习就装windows版本，方便测试
- https://www.rabbitmq.com官网，下载windows版本。需要先安装一个erlang的环境。
- 先安装环境，后安装rabbitmq.一直next即可。
- 打开rabbitmq命令行，然后输命令
rabbitmq-plugins enable rabbitmq_management
- 浏览器打开 http://localhost:15672/ 账号/密码都是guest
1.3 用户管理
image.png
1.4 virtual hosts
virtual hosts 相当于数据库的db
1.先新增
点击
2.点击新增，进行配置。增加一个用户
image.png
二 简单队列
模型
image.png
P：消息的生产者
红色的：消息队列
C：消费者
2.1 创建项目
1.用idea创建一个项目，添加依赖
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.zyc</groupId>
<artifactId>myrabbit1</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.6.0</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.10</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.5</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>
2.2 获取连接
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author zhuyc
* @create 2019-05-08 22:02
*/
public class ConnectiionUtil {
public static Connection getConnection() throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
factory.setPort(5672);
factory.setVirtualHost("/vhost_zhuyc");
factory.setUsername("zhuyc");
factory.setPassword("123456");
return factory.newConnection();
}
}
2.3 发送消息
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.zhuyc.util.ConnectiionUtil;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author zhuyc
* @create 2019-05-08 22:15
*/
public class Sender {
private static final String QUEUE_NAME = "test_simple_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//获取连接
Connection connection = ConnectiionUtil.getConnection();
//从连接中获取一个通道
Channel channel = connection.createChannel();
//创建队列声明
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
String msg = "hello world!simple queue";
channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
System.out.println("send end");
channel.close();
connection.close();
}
}
2.4 消息接收
import com.rabbitmq.client.*;
import com.zhuyc.util.ConnectiionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author zhuyc
* @create 2019-05-08 22:27
*/
public class Receiver
{
private static final String QUEUE_NAME = "test_simple_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//获取连接
Connection connection = ConnectiionUtil.getConnection();
//从连接中获取一个通道
Channel channel = connection.createChannel();
DefaultConsumer consumer = new DefaultConsumer(channel) {
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
// 接收到的消息
String message = new String(body);
System.out.println("Receive 接收到的消息 " + message);
}
};
channel.basicConsume(QUEUE_NAME,true,consumer);
}
}
这里视频中的rabbitMQ client的版本较低，和新版的api不一样。需要看对应版本的官方文档
参考：https://blog.csdn.net/qq_37653556/article/details/82753890
2.5 简单队列的不足
耦合性高，生产者一一对应消费者（如果想要多个消费者消费队列中的消息，这时候就不行了）。
队列名变更。这时候得同时变更
三 Work queues 工作队列
模型
image.png
代码和上面是一样，只是多了一个消费者去监听队列
2.1 轮询分发
消费者1
package com.zhuyc.work;
import com.rabbitmq.client.*;
import com.zhuyc.util.ConnectiionUtil;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author zhuyc
* @create 2019-05-11 16:24
*/
public class Receiver1 {
private static final String QUEUE_NAME = "test_work_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//获取连接
Connection connection = ConnectiionUtil.getConnection();
//从连接中获取一个通道
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
DefaultConsumer consumer = new DefaultConsumer(channel) {
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
// 接收到的消息
String message = new String(body);
System.out.println("Receive1 接收到的消息 " + message);
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("Receive1 done" );
}
}
};
channel.basicConsume(QUEUE_NAME,true,consumer);
}
}
消费者2
public static void main(String[] args) throws IOException, TimeoutException {
//获取连接
Connection connection = ConnectiionUtil.getConnection();
//从连接中获取一个通道
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
DefaultConsumer consumer = new DefaultConsumer(channel) {
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
// 接收到的消息
String message = new String(body);
System.out.println("Receive2 接收到的消息 " + message);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("Receive2 done" );
}
}
};
现象
消费者1和消费者处理得消息数量是一样的，你一个我一个
这种方式叫做轮询分发，就是不管哪个消费者效率较高，处理速度较快，平均分
2.2 公平分发
使用basicQos（prefetch=1）,即每次取完之后会对消息队列反馈。（注意：使用公平分发，必须关闭自动应答ack，改成手动）
生产者
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
Connection connection = ConnectiionUtil.getConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
/**
* 每个消费者发送确认消息之前，消息队列不发送下一条消息到消费者，一次只处理一个消息
*
* 限制发送给同一个消费者 不得超过一条消息
*/
int prefetchCount = 1;
channel.basicQos(prefetchCount);
for (int i =0;i<50;i++){
String msg = "hello"+i;
channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
Thread.sleep(30);
System.out.println(i+" 发送完毕");
}
System.out.println("send end..");
}
消费者
public static void main(String[] args) throws IOException, TimeoutException {
//获取连接
Connection connection = ConnectiionUtil.getConnection();
//从连接中获取一个通道
final Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
channel.basicQos(1);//保证一次只分发一个
DefaultConsumer consumer = new DefaultConsumer(channel) {
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
// 接收到的消息
String message = new String(body);
System.out.println("Receive1 接收到的消息 " + message);
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("Receive1 done" );
//手动回执
channel.basicAck(envelope.getDeliveryTag(),false);
}
}
};
boolean autoAck = false;//关闭自动应答ack
channel.basicConsume(QUEUE_NAME,autoAck ,consumer);
现象
这样消费快的消费者就会消费更多的消息
参考
1.视频讲解
