RabbitMQ快速入手(docker版)

1.docker安装

ps：rabbitmq:management是带图形化界面的，在linux环境下不用安装elang环境 5672是rabbitmq的端口类似3306，15672标识图形化界面的端口

docker run -d -p 5672:5672 \
-p 15672:15672 \
--name rabbitmq \
rabbitmq:management

访问：虚拟机ip+端口号。默认账号密码guest

2.图形化界面介绍

在了解图形化界面之前需要先了解RabbitMQ的执行流程

Connection就相当于java端或者go端创建连接

​​

2.1Overview

2.2Connections

2.3Channels

2.4Exchanges

2.5Queues

2.6Admin

这个页面主要分为6的小点

administrator 可以登录控制台，查看所有信息，可以对rabbitmq进行管理，消息的提供者
monitoring 监控者，可以登录和查看信息，不能管理，消息的消费者
policymaker 策略制定者 登录控制台
managment 普通管理员、登录控制台

2.6.1users

No access表示未分配虚拟机

2.6.2Virtual Hosts

每一个虚拟主机都有一整套的exchang和queue,方便rabbitmq做不同的业务

2.6.3Feature Flags

比较高级，暂时还没研究到

2.6.4Policies

消息策略

2.6.5Limits

规定rabbitmq当前虚拟机能最大连接和队列最大多少，生产环境还是要设置一下的,正常几十个就行

2.6.2Cluster

规定当前结点名字

3.命令行控制

rabbitmqctl --help 查看所有指令

状态查看：

rabbitmqctl status 查看状态

rabbitmqctl list_bindings 查看绑定

rabbitmqctl list_channels 查看channel

rabbitmqctl list_connectios 查看connectios

rabbitmqctl list_consumers 查看消费者

rabbitmqctl list_exchanges 查看交换机

队列相关：

rabbitmqctl list_queues 查看队列

rabbitmqctl delete_queue 队列名字 删除队列

rabbitmqctl purge_queue 队列名字 清空队列

用户相关：

rabbitmgctl add user 新建用户：

rabbitmqctl change_password 修改用户密码：

rabbitmgctl delete user 删除用户

rabbitmgctl list users 查看用户

rabbitmqctl rabbitmgctl set_user_tags 设置用户角色

应用启动：

rabbitmqctl start_app 启动应用
rabbitmqctl stop_app 关闭应用保留Erlang虚拟机（暂停）
rabbitmqctl stop 关闭应用并关闭Erlang虚拟机

集群相关：

rabbitmqctl join_cluster 加入集群

rabbitmqctl reset 离开集群

镜像队列：

rabbitmqctl sync_queue 设置镜像队列

rabbitmqctl cancel_sync_queue 取消镜像队列

4.消息队列使用注意事项

消息流转规则：

消息能直接将消息发送给最终队列，必须发送到交换机，再有交换机发送到队列中

数量设置:

1.交换机数量不能过多，一般同一个业务或同一类业务使用同一个交换机

2.合理设置队列数量，一个微服务监听一个队列，或一个微服务的一个物业监听一个队列

3.合理配置交换机类型，使用Topic模式时仔细设置绑定值

尽量时候自动化配置：

1.讲交换机/队列的操作固化在应用代码中，减少运维操作

2.交换机由双方同时声明，队列由接收方声明并配置绑定关系

3.交换机/队列的参数由双方团队确认，否则重复声明，如果参数不一致会导致声明失效

4.同步和异步

** 异步收发**

创建消费者对象，接受不到消息，消费者对象挂载到后台运行，等待（阻塞）状态。有消息发送立刻激活。

同步收发

在移动端创建定时器，然后向后端java项目发出轮询请求，后端java项目接受到轮询请求之后，用同步方式接受队列中的消息，然后把消息队列存储到MongoDB中，最后向小程序返回接收了多少条新消息

ps.RMQ比较其他MQ产品更可靠，最快的是（卡夫卡）

一般依靠多线程，java代码异步，RabitMQ同步

5.RabbitMQ工作模式

** 1）简单模式**

一个生产者（发送方）对应一个消费者（接收方）1对1

** 2）工作（work）模式**

一个生产者对应多个消费者，但是只能由一个额消费者获得消息（排他）

举例：

** 3）交换机模式**

发布/订阅模式

一个消费者将消息首先发送到fanout交换器，交换器绑定多个队列，然后与对应的所有消费者都能接受信息（不排他）1对多，无条件转发

** 路由模式**

生产者将消息发送到direct交换器，交换器按照关键字（Key）,把消息路由到某个队列

1对多，但是有转发规则(附带关键字)，符合才转发

** 4）主题模式**

生产者将消息发送到Topic交换器，交换器按照负责的规则（写表达式，大于，小于等），把消息路由到某个队列

可以实现1对1，也可以实现1对多

6.maven连接RabbitMQ小案例

6.1创建连接

创建maven工程

导入依赖

    <dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.12.0</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.32</version>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>3.12.0</version>
        </dependency>
    </dependencies>

配置logfj

log4j.rootLogger=DEBUG,A1 log4j.logger.com.taotao = DEBUG 
log4j.logger.org.mybatis = DEBUG
log4j.appender.A1=org.apache.log4j.ConsoleAppender
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
log4j.appender.A1.layout.ConversionPattern=%-d{yyyy-MM-dd HH:mm:ss,SSS} [%t] [%c]-[%p] %m%n

创建连接

package com.mq.utils;

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 {
        //    1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
//    在工厂对象中设置MQ的连接信息（(ip,port,virtualhost,username,password)
        factory.setHost("hadoop102");
        factory.setPort(5672);
        factory.setVirtualHost("host1");
        factory.setUsername("qing");
        factory.setPassword("123456");
        //通过工厂对象获取mq的连接
        Connection connection = factory.newConnection();
        return connection;

    }
    public static void main(String[] args) throws IOException, TimeoutException {
        System.out.println(getConnection());
    }
}

结果：

6.2简单模式demo

按照上面的工作图，一个发送者，一个接收者

发送者

package com.mq.service;

import com.mq.utils.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

public class SendMsg {
    public static void main(String[] args) throws IOException, TimeoutException {
        String msg = "helloword1";
        Connection connection = ConnectionUtil.getConnection();//相当于数据库jdbc连接
        Channel channel = connection.createChannel();//相当于JDBc中的statement

        //定义队列
        //参数1：定义队列名称
        // 2.是否持久化
        // 3.是否排外（当连接关闭是，是否删除这个队列）
        // 4.自动删除（当此队列的连接数为0的时候，此队列会销毁，无论队列是否还有数据）
        //5.设置当前队列的参数
//        channel.queueDeclare("queue7",false,false,false,null);
        //参数
//        1.交换机名称，如果直接发送到消息队列，则交换机名称为“”空
//        2.目标队列名称
        //3.设置当前这条消息的时间
        //4.参数4.消息的内容
        //ps：如果是路由模式 设置交换机后，第二个参数标识为key，而不队列
        channel.basicPublish("", "Queue1",null ,msg.getBytes());
        System.out.println("发送"+msg);
        channel.close();
        connection.close();

    }

}

接收者

import com.mq.utils.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ReceiveMsg {
    public static void main(String[] args) throws IOException, TimeoutException {
//        建立连接
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        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("msg = " + msg);
            }
        };
//        参数1表示从哪个队列获取数据，参数二：是否应答，获取数据得到反馈，参数3：获取数据
        channel.basicConsume("Queue1",true,consumer);
        channel.close();
        connection.close();

    }
}

11.在springboot中使用MQ

创建两个服务一个发送，一个接受

导入依赖

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-thymeleaf</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>

yml配置文件

server:
  port: 8081
spring:
  rabbitmq:
    host: hadoop102
    port: 5672
    virtual-host: host1
    username: qing
    password: 123456

发送方

ps:实现发送主要使用AMqpTemplate模板注入

@Service
public class TestService {
    @Autowired
    private AmqpTemplate amqpTemplate;
    public void sendMsg(String msg){
        if(msg.startsWith("s1")){
            //参数1队列，参数2信息
            amqpTemplate.convertAndSend("queue1",msg);
        }else if (msg.startsWith("s2")){
            //将消息发送交换机
            //第二个参数不给表示参数1是交换机,参数2设置为空（订阅/发布）
            amqpTemplate.convertAndSend("ex1","",msg);
        }else if(msg.startsWith("s3")){
            //3.发送消息到交换机（路由模式）
            amqpTemplate.convertAndSend("ex2","a",msg);
        }

    }
}

接收方：

通过@RabbitListener注解实现，表示接受的队列，@RabbitHandler表示接受到的数据

@Service
//接受数据可以配置多个队列
@RabbitListener(queues = {"queue1"})
public class ReceiveService {

    //接受的数据传到@RabbitHandler的方法
    @RabbitHandler
//    注意接受类型需要跟传递的参数类型一致
    public void receiveMsg(String msg){
        System.out.println("接受MSG " + msg);

    }

}

7.使用RabbtMQ传输对象流消息

消息队列可以发送，字符串、字节数组、序列化（Serizarable）对象.implements Serializable.

只有序列化对象才可以进行网络传输，用对象传，对象接并且传递对象的包名，类名、属性名必须一致。

传输的三种方式：

方式1.在实体类上直接实现implements Serializable结构

方式2：进行传byte通过SerializationUtils手动转

消息提供者：byte[] bytes=SerializationUtils.serialize(对象)

消息接收者：SerializationUtils.deserialize(对象)

方式3：使用JSON字符串传递（要求对象名一致）

消息提供者：

    ObjectMapper object=new ObjectMapper();

    String msg=objectMapper.writeValueAsString(对象);

消息接收者：

 ObjectMapper objectMapper  = new ObjectMapper();

    * 对象=objectMapper.readValue(msg,*.class)

8.基于java的交换机与队列的创建

消息队列和交换机可以通过web管理系统创建和修改，也可以通过java代码创建完成

9普通maven项目项目交换机创建

Connection connection=ConnectionUtil.getConnection();

Channel channel=connection.createChannel();

创建队列

channel.queueDeclare("队列名"，”是否持久化“，”是否排外"，“自动删除"，”设置当前队列参数“)

举例：channel.queueDeclare("queueq"，”false“，”false"，“false"，”null“)

创建交换机

channel.excahngeDeclare("交换器名"，类型)

举例:

创建订阅交换机channel.excahngeDeclare("交换器名"，BuiltinExchangeType.FANOUT)

创建路由交换机：channel.excahngeDeclare("交换器名"，BuiltinExchangeType.DIRECT)

绑定交换机

channel.queueBind("队列名"，"交换机"，key);

13.2Springboot通过配置

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {
    //新建队列，创建多个队列
    @Bean("q8")
    public Queue newQueue8() {

        Queue queue8 = new Queue("queue8");
        //可以设置属性
        queue8.isAutoDelete();
        return new Queue("queue8");
    }

    @Bean("q9")
    public Queue newQueue9() {

        Queue queue8 = new Queue("queue8");
        //可以设置属性
        queue8.isAutoDelete();
        return new Queue("queue8");
    }

    //声名交换机
    @Bean
    public FanoutExchange newFanoutExchange() {
        return new FanoutExchange("ex3");
    }

    //声名路由模式交换机
    @Bean
    public DirectExchange newDirectExchange() {
        return new DirectExchange("ex4");
    }

    //绑定队列 ps:这个队列穿的形参名字需要是创建队列的bean的名字
    @Bean
    public Binding bindingDirect9(Queue q8,Queue q9,DirectExchange m) {
        return BindingBuilder.bind(q8).to(m).with("k1");
    }
    @Bean
    public Binding bindingDirect10(Queue q8,Queue q9,DirectExchange m) {
        return BindingBuilder.bind(q9).to(m).with("k1");
    }

}

10.MQ-maven应用消息确定-消息可靠性

10.1开启事务

但在发送消息冲添加事务，处理效率会降低几十倍，一般不开启

开启事务

channel.txSelect();//开启事务

try{

//mq代码

}catch(Exception e){

channel.txRollback;//事务回滚

}

10.2消息确认和return机制

消息确认：生产者->交换机

retun机制：交换机->队列

消息确认机制

        channel.confirmSelect();//发送消息前开启消息队列

        boolean b = channel.waitForConfirms();//接受消息确认消息

但是一个线程如果批量发送消息，但线程会卡，所以加一个监听，多开一个线程帮助查看返回结果

        channel.addConfirmListener(new ConfirmListener() {
//            成功响应
//参数1：返回数据的表示
//参数2：是否为批量，一条为true，多条为false
            @Override
            public void handleAck(long l, boolean b) throws IOException {

            }
//失败响应
            @Override
            public void handleNack(long l, boolean b) throws IOException {

            }
        });

return机制

        //设置retun监听器
        channel.addReturnListener(new ReturnListener() {
            //1.返回表示2.3.交换机的名字4.交换机对应的队列key5.发送的消息
            @Override
            public void handleReturn(int i, String s, String s1, String s2, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {
                //如果交换机分发到队列失败
            }
        });
        //第三个参数只有开启return机制才可以，true表示开启见监听
        channel.basicPublish("ex1", "e",true,null ,msg.getBytes());

12.SpringBoot的消息确认和retun机制

15.1在application.yml开启消息确认和return机制

spring:
  rabbitmq:
    host: hadoop102
    port: 5672
    virtual-host: host1
    username: qing
    password: 123456
#    开启消息确认
    publisher-confirm-type: simple
#    开启rentun机制
    publisher-returns: true

编写配置文件

@Component
public class MsgConirmAndreturn implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    Logger logger= (Logger) LoggerFactory.getLogger(MsgConirmAndreturn.class);
    //AMQP表示通信协议AmqpTemplate相当于对rabbitTemplate的封装
    @Resource
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init(){
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

    //方法用于监听消息确认的结果
    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
    }

    //当交换机分发消息失败时执行
    @Override
    public void returnedMessage(Message message, int i, String s, String s1, String s2) {

    }
}

13.MQ死信队列实现消息延迟

延迟队列消息进入到队列中，延迟指定的时间才能被消费者消费

AMQP歇息和RabbitMQ本身是不支持RabbitMQ延迟队列功能，但是可以通过ttl特性延迟队列功能

16.1死信队列的创建

要求：1.消息没有人消费2.消息有延迟时间

创建交换机

并在交换机创建响应队列

创建队列

实现：往k1的队列发送，从k2的队列接受

14.应用场景举例

解耦

订单系统和库存系统，传统调用方法比如OpenFeign,但这属于远程调用，而消息队列是同通知你去修改

异步

用户注册需要注册邮件和短信

聊天室

缓存作用