第二十一章 RabbitMQ

1.2 RabbitMQ 简介

[root@localhost ~]# uname -r
3.10.0-1160.el7.x86_64

[root@localhost ~]# yum -y update

[root@localhost ~]# yum remove docker  docker-common docker-selinux docker-engine

[root@localhost ~]# yum install -y yum-utils device-mapper-persistent-data lvm2

[root@localhost ~]# yum-config-manager --add-repo http://download.docker.com/linux/centos/docker-ce.repo（中央仓库）
[root@localhost ~]# yum-config-manager --add-repo http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo（阿里仓库）

[root@localhost ~]# yum list docker-ce --showduplicates | sort -r

[root@localhost ~]# yum -y install docker-ce-18.03.1.ce

[root@localhost ~]# systemctl start docker
[root@localhost ~]# systemctl enable docker 设置开机自启

[root@localhost ~]# docker version

[root@localhost ~]# docker search rabbitmq:management

[root@localhost ~]# docker pull macintoshplus/rabbitmq-management

[root@localhost ~]# docker images

[root@localhost ~]# systemctl start docker
[root@localhost ~]# systemctl enable docker  设置开机自启

[root@localhost ~]# docker ps -a

三、RabbitMQ 六种消息模型

3.1 基本消息模型

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

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ConnectionUtil {public static Connection getConnection() throws IOException, TimeoutException {// 定义连接工厂ConnectionFactory factory = new ConnectionFactory();// 设置服务器地址factory.setHost("192.168.111.133");// 设置端口factory.setPort(5672);/***  设置账号信息，用户名、密码、vhost*  设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq*//*factory.setVirtualHost("/ly");factory.setUsername("ly");factory.setPassword("123456")*/;// 通过工厂获取连接Connection connection = factory.newConnection();return connection;}}

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Send {private  final static String QUEUE_NAME = "simple_queue";public static void main(String[] args) throws IOException, TimeoutException {// 1. 获取到连接Connection connection = ConnectionUtil.getConnection();// 2. 从连接中创建通道，使用通道才能完成消息相关的操作Channel channel = connection.createChannel();// 3. 声明（创建）队列// 参数：String queue，boolean durable，boolean exclusive，boolean autoDelete, Map<String, Object> arguments/*** 参数明细：* 1.queue，队列名称* 2.durable，是否持久化，如果持久化，mq重启后队列还在* 3.exclusive，是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除，*              如果将此参数设置为true可用于临时队列的创建* 4.autoDelete，自动删除，队列不再使用时是否自动删除此队列，*              如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）* 5.arguments，参数，可以设置一个队列的扩展参数，比如：可设置存活时间*/channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 消息内容String message = "Hello World!";// 向指定的队列中发送消息//参数：String exchange, String routingKey, BasicProperties props, byte[] body/*** 参数明细：* 1.exchange，交换机，如果不指定将使用mq的欧仁交换机（设置为"")* 2.routingKey，路由Key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名称* 3.props，消息的属性* 4.body，消息的内容*/channel.basicPublish("", QUEUE_NAME, null, message.getBytes());System.out.println("[x] Send '" + message + "'");// 关闭通道和连接（资源关闭最好用try-catch-finally语句处理）channel.close();connection.close();}}

import com.ly.rabbitmq.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Recv {private final static String QUEUE_NAME = "direct_exchange_queue_sms";//短信队列private final static String EXCHANGE_NAME = "test_direct_exchange";public static void main(String[] argv) throws IOException, TimeoutException {/* 获取到连接 */Connection connection = ConnectionUtil.getConnection();// 获取通道Channel channel = connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 绑定队列到交换机，同时指定需要订阅的routing key。可以指定多个channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "sms");//指定接收发送方指定routing key为sms的消息//channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "email");// 定义队列的消费者DefaultConsumer consumer = new DefaultConsumer(channel) {// 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {// body 即消息体String msg = new String(body);System.out.println(" [短信服务] received : " + msg + "!");}};// 监听队列，自动ACKchannel.basicConsume(QUEUE_NAME, true, consumer);}
}

import com.ly.rabbitmq.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Recv2 {private final static String QUEUE_NAME = "direct_exchange_queue_email";//邮件队列private final static String EXCHANGE_NAME = "test_direct_exchange";public static void main(String[] argv) throws IOException, TimeoutException {// 获取到连接Connection connection = ConnectionUtil.getConnection();// 获取通道Channel channel = connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 绑定队列到交换机，同时指定需要订阅的routing key。可以指定多个channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "email");//指定接收发送方指定routing key为email的消息// 定义队列的消费者DefaultConsumer consumer = new DefaultConsumer(channel) {// 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {// body 即消息体String msg = new String(body);System.out.println(" [邮件服务] received : " + msg + "!");}};// 监听队列，自动ACKchannel.basicConsume(QUEUE_NAME, true, consumer);}
}

3.5 Topics 通配符模式（交换机类型：topics）

import com.ly.rabbitmq.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Send {private final static String EXCHANGE_NAME = "test_topic_exchange";public static void main(String[] argv) throws IOException, TimeoutException {// 获取到连接Connection connection = ConnectionUtil.getConnection();// 获取通道Channel channel = connection.createChannel();// 声明exchange，指定类型为topicchannel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);// 消息内容String message = "这是一只行动迅速的橙色的兔子";// 发送消息，并且指定routing key为：quick.orange.rabbitchannel.basicPublish(EXCHANGE_NAME, "quick.orange.rabbit", null, message.getBytes());System.out.println(" [动物描述：] Sent '" + message + "'");channel.close();connection.close();}
}

import com.ly.rabbitmq.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Recv {private final static String QUEUE_NAME = "topic_exchange_queue_Q1";private final static String EXCHANGE_NAME = "test_topic_exchange";public static void main(String[] argv) throws IOException, TimeoutException {// 获取到连接Connection connection = ConnectionUtil.getConnection();// 获取通道Channel channel = connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 绑定队列到交换机，同时指定需要订阅的routing key。订阅所有的橙色动物channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.orange.*");// 定义队列的消费者DefaultConsumer consumer = new DefaultConsumer(channel) {// 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {// body 即消息体String msg = new String(body);System.out.println(" [消费者1] received : " + msg + "!");}};// 监听队列，自动ACKchannel.basicConsume(QUEUE_NAME, true, consumer);}
}

import com.ly.rabbitmq.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;public class Recv2 {private final static String QUEUE_NAME = "topic_exchange_queue_Q2";private final static String EXCHANGE_NAME = "test_topic_exchange";public static void main(String[] argv) throws Exception {// 获取到连接Connection connection = ConnectionUtil.getConnection();// 获取通道Channel channel = connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 绑定队列到交换机，同时指定需要订阅的routing key。订阅关于兔子以及懒惰动物的消息channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.*.rabbit");channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "lazy.＃");// 定义队列的消费者DefaultConsumer consumer = new DefaultConsumer(channel) {// 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {// body 即消息体String msg = new String(body);System.out.println(" [消费者2] received : " + msg + "!");}};// 监听队列，自动ACKchannel.basicConsume(QUEUE_NAME, true, consumer);}
}

3.6 RPC