第二十一章 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) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [短信服务] received : " + msg + "!");
            }
        };
        // 监听队列，自动ACK
        channel.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) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [邮件服务] received : " + msg + "!");
            }
        };
        // 监听队列，自动ACK
        channel.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，指定类型为topic
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        // 消息内容
        String message = "这是一只行动迅速的橙色的兔子";
        // 发送消息，并且指定routing key为：quick.orange.rabbit
        channel.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) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public 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 + "!");
            }
        };
        // 监听队列，自动ACK
        channel.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) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public 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 + "!");
            }
        };
        // 监听队列，自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

3.6 RPC