RabbitMQ基础与实操复习

这篇具有很好参考价值的文章主要介绍了RabbitMQ基础与实操复习。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

官网有七种模型,这里仅介绍五种

1、MQ引言

1.1 什么是MQ

  MQ(Message Quene) : 翻译为消息队列,通过典型的 生产者消费者模型,生产者不断向消息队列中生产消息,消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的,而且只关心消息的发送和接收,没有业务逻辑的侵入,轻松的实现系统间解耦。别名为 消息中间件通过利用高效可靠的消息传递机制进行平台无关的数据交流,并基于数据通信来进行分布式系统的集成。

1.2 MQ有哪些

  当今市面上有很多主流的消息中间件,如老牌的ActiveMQRabbitMQ,炙手可热的Kafka,阿里巴巴自主开发RocketMQ等。

1.3 不同MQ特点

# 1.ActiveMQ
		ActiveMQ 是Apache出品,最流行的,能力强劲的开源消息总线。它是一个完全支持JMS规范的的消息中间件。丰富的API,多种集群架构模式让ActiveMQ在业界成为老牌的消息中间件,在中小型企业颇受欢迎!

# 2.Kafka
		Kafka是LinkedIn开源的分布式发布-订阅消息系统,目前归属于Apache顶级项目。Kafka主要特点是基于Pull的模式来处理消息消费,
		追求高吞吐量,一开始的目的就是用于日志收集和传输。0.8版本开始支持复制,不支持事务,对消息的重复、丢失、错误没有严格要求,
		适合产生大量数据的互联网服务的数据收集业务。

# 3.RocketMQ
		RocketMQ是阿里开源的消息中间件,它是纯Java开发,具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。RocketMQ思路起
		源于Kafka,但并不是Kafka的一个Copy,它对消息的可靠传输及事务性做了优化,目前在阿里集团被广泛应用于交易、充值、流计算、消
		息推送、日志流式处理、binglog分发等场景。

# 4.RabbitMQ
		RabbitMQ是使用Erlang语言开发的开源消息队列系统,基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由(包括点对点和
		发布/订阅)、可靠性、安全。AMQP协议更多用在企业系统内对数据一致性、稳定性和可靠性要求很高的场景,对性能和吞吐量的要求还在
		其次。
		

   RabbitMQ比Kafka可靠,Kafka更适合IO高吞吐的处理,一般应用在大数据日志处理或对实时性(少量延迟),可靠性(少量丢数据)要求稍低的场景使用,比如ELK日志收集。

2、RabbitMQ引言

2.1 RabbitMQ

  基于AMQP协议,erlang语言开发,是部署最广泛的开源消息中间件,是最受欢迎的开源消息中间件之一。

RabbitMQ基础与实操复习

官网: https://www.rabbitmq.com/

官方教程: https://www.rabbitmq.com/#getstarted

AMQP协议:

  AMQP(advanced message queuing protocol)`在2003年时被提出,最早用于解决金融领不同平台之间的消息传递交互问题。顾名思义,AMQP是一种协议,更准确的说是一种binary wire-level protocol(链接协议)。这是其和JMS的本质差别,AMQP不从API层进行限定,而是直接定义网络交换的数据格式。这使得实现了AMQP的provider天然性就是跨平台的。以下是AMQP协议模型:

RabbitMQ基础与实操复习

2.2 RabbitMQ安装

  这个以前写过,不重复介绍了 https://codeleader.blog.csdn.net/article/details/121890708

3、RabbitMQ配置

3.1 RabbitMQ命令行

# 1.服务启动相关
	systemctl start|restart|stop|status rabbitmq-server

# 2.管理命令行  用来在不使用web管理界面情况下命令操作RabbitMQ
	rabbitmqctl  help  可以查看更多命令

# 3.插件管理命令行
	rabbitmq-plugins enable|list|disable 

3.2 Web管理界面

3.2.1 overview概览

RabbitMQ基础与实操复习

  • connections:无论生产者还是消费者,都需要与RabbitMQ建立连接后才可以完成消息的生产和消费,在这里可以查看连接情况

  • channels:通道,建立连接后,会形成通道,消息的投递获取依赖通道。

  • Exchanges:交换机,用来实现消息的路由

  • Queues:队列,即消息队列,消息存放在队列中,等待消费,消费后被移除队列。

3.2.2 Admin用户和虚拟主机管理

1、添加用户

RabbitMQ基础与实操复习

上面的Tags选项,其实是指定用户的角色,可选的有以下几个:

  • 超级管理员(administrator)

    可登陆管理控制台,可查看所有的信息,并且可以对用户,策略(policy)进行操作。

  • 监控者(monitoring)

    可登陆管理控制台,同时可以查看rabbitmq节点的相关信息(进程数,内存使用情况,磁盘使用情况等)

  • 策略制定者(policymaker)

    可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。

  • 普通管理者(management)

    仅可登陆管理控制台,无法看到节点信息,也无法对策略进行管理。

  • 其他

    无法登陆管理控制台,通常就是普通的生产者和消费者。

    2、创建虚拟主机

    RabbitMQ基础与实操复习

默认创建虚拟主机之后,没有用户可以使用,需要绑定用户

3、绑定虚拟主机和用户

创建好虚拟主机,我们还要给用户添加访问权限:

点击添加好的虚拟主机:

RabbitMQ基础与实操复习

这里给admin和ems都授权,授权之后就能在页面上看到了,如下图

RabbitMQ基础与实操复习

4、RabbitMQ常用消息模型测试

4.1 RabbitMQ支持的消息模型

RabbitMQ基础与实操复习

RabbitMQ基础与实操复习

4.2 引入依赖

<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.16.0</version>
</dependency>

4.3 第一种模型:直连

RabbitMQ基础与实操复习

在上图的模型中,有以下概念:

  • P:生产者,也就是要发送消息的程序
  • C:消费者:消息的接受者,会一直等待消息到来。
  • queue:消息队列,图中红色部分。类似一个邮箱,可以缓存消息;生产者向其中投递消息,消费者从其中取出消息。

4.3.1 自定义连接工具类

ublic class RabbitUtils {
    //创建连接MQ的连接工厂   重量级资源
    public static ConnectionFactory connectionFactory=new ConnectionFactory();
    static {    //类加载执行  只执行一次
        //设置连接rabbitmq主机
        connectionFactory.setHost("ip");
        //设置端口号
        connectionFactory.setPort(5672);
        //设置连接哪个虚拟主机
        connectionFactory.setVirtualHost("/ems");
        //设置访问虚拟主机的用户名和密码
        connectionFactory.setUsername("ems");
        connectionFactory.setPassword("密码");
    }

    //定义提供连接对象的方法
    public static Connection getConnection() {
        Connection connection = null;
        try {

            //获取连接对象
            connection = connectionFactory.newConnection();
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
        return connection;
    }

    //关闭通道和关闭连接的方法
    public static void closeConnectionAndChannel(Channel channel, Connection connection) {
        try {
            if (channel != null) {
                channel.close();
            }
            if (connection != null) {
                connection.close();
            }
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
    }
}

4.3.2 生产者

public class Provider {

    //生产消息   HelloWorld:直连模式

    public static void main(String[] args) throws IOException {
  
       Connection connection= RabbitUtils.getConnection();
        //获取连接中的通道
        Channel channel = connection.createChannel();
        //通道绑定对应的消息队列
        //参数1:队列名称,如果不存在,自动创建。
        //参数2:定义队列特性是否持久化 true :持久化,false:不持久化
        //参数3:是否独占队列
        //参数4:是否在消费完成后自动删除队列 true:自动删除,false:不自动删除
        //参数5:额外附加参数
        channel.queueDeclare("hello",false,false,false,null);
        //发布消息
        //交换机名称,队列名称,传递消息的额外设置,消息的具体内容
        channel.basicPublish("","hello",null,"hello rabbitmq".getBytes());
        RabbitUtils.closeConnectionAndChannel(channel,connection);
    }
}

  这里不指定交换机名称,用的就是默认交换机。

4.3.3 消费者

public class Consumer {

    public static void main(String[] args) throws IOException, TimeoutException {
       //调用自定义工具类
        Connection connection= RabbitUtils.getConnection();

        //创建通道
        Channel channel = connection.createChannel();
        //通道绑定对象
        channel.queueDeclare("hello",true,false,true,null);

        //消费消息
        //参数1:消费哪个队列的消息
        //参数2:开启消息的自动确认机制
        //参数3:消费消息时的回调接口
        channel.basicConsume("hello",true,new DefaultConsumer(channel){
            //body:消息队列中取出的消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("=============="+new String(body));
            }
        });

        //调用工具类
        RabbitUtils.closeConnectionAndChannel(channel,connection);

    }
}

启动生产者:

RabbitMQ基础与实操复习

启动消费者:

RabbitMQ基础与实操复习

可以看到,消息已经收到了

4.4 第二种模型:Work Queue

  Work queues,也被称为(Task queues),任务模型。当消息处理比较耗时的时候,可能生产消息的速度会远远大于消息的消费速度。长此以往,消息就会堆积越来越多,无法及时处理。此时就可以使用work 模型:让多个消费者绑定到一个队列,共同消费队列中的消息。队列中的消息一旦消费,就会消失,因此任务是不会被重复执行的。

RabbitMQ基础与实操复习

角色:

  • P:生产者:任务的发布者
  • C1:消费者-1,领取任务并且完成任务,假设完成速度较慢
  • C2:消费者-2:领取任务并完成任务,假设完成速度快

4.4.1 生产者:

public class Provider {
    public static void main(String[] args) throws IOException {
        //获取连接对象
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();
        //通过通道声明队列
        channel.queueDeclare("work",true,false,false,null);
        //生产消息
        for (int i = 1; i <=20 ; i++) {
            channel.basicPublish("","work", null,(i+" hello work queue").getBytes());
        }

        //关闭资源
        RabbitUtils.closeConnectionAndChannel(channel,connection);
    }
}

4.4.2 消费者

消费者1:

//轮询分发测试
public class Consumer1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare("work",true,false,false,null);

        channel.basicConsume("work",true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者-1:"+new String(body));
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

消费者2:

public class Consumer2 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare("work",true,false,false,null);
        //参数1:队列名称,参数2:消息自动确认 true:消费者自动向rabbitmq确认消息消费了,false:不会自动确认消息
        channel.basicConsume("work",true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者-2:"+new String(body));
            }
        });
    }
}

先启动两个消费者

RabbitMQ基础与实操复习

再启动生产者:

RabbitMQ基础与实操复习

RabbitMQ基础与实操复习

   总结:默认情况下,RabbitMQ将按顺序将每个消息发送给下一个使用者。平均而言,每个消费者都会收到相同数量的消息。这种分发消息的方式称为循环。

   可以看到,默认是轮询分发的,但是这样子不好,我们的消费者1使用线程休眠了1s处理的很慢依然和消费者2五五开。

   我们想要的结果是能者多劳,也就是处理速度快的就尽量多处理几条消息。

改进如下:

  • 设置一次只接受一条未确认的消息
  • 关闭消息自动确认,改为手动确认

4.4.3 改进为能者多劳

生产者不动,改变消费者

消费者1:

//能者多劳测试
public class Consumer1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();

        channel.basicQos(1);//每次只能消费一个消息
        channel.queueDeclare("work",true,false,false,null);

        channel.basicConsume("work",false,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者-1:"+new String(body));
                //手动确认
                //参数1:手动确认消息标识, 参数2:false 每次确认一个
                channel.basicAck(envelope.getDeliveryTag(),false);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

消费者2:

public class Consumer2 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();

        channel.basicQos(1);//每次只能消费一个消息
        channel.queueDeclare("work",true,false,false,null);
        //参数1:队列名称,参数2:消息自动确认 true:消费者自动向rabbitmq确认消息消费了,false:不会自动确认消息
        channel.basicConsume("work",false,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者-2:"+new String(body));
                //手动确认
                //参数1:手动确认消息标识, 参数2:false 每次确认一个
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        });
    }
}

消费者2:

RabbitMQ基础与实操复习

消费者2:

RabbitMQ基础与实操复习

可以看到,达到了能者多劳的效果

4.5 第三种模型:Fanout

fanout 扇出 也称为广播

RabbitMQ基础与实操复习

RabbitMQ基础与实操复习

在广播模式下,消息发送流程是这样的:

  • 可以有多个消费者
  • 每个消费者有自己的queue(队列)
  • 每个队列都要绑定到Exchange(交换机)
  • 生产者发送的消息,只能发送到交换机,交换机来决定要发给哪个队列,生产者无法决定。
  • 交换机把消息发送给绑定过的所有队列
  • 队列的消费者都能拿到消息。实现一条消息被多个消费者消费

4.5.1 生产者

public class Provider {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();
        //将通道声明指定的交换机
        //参数1:交换机名称,参数2:交换机类型,fanout:广播类型
        channel.exchangeDeclare("logs","fanout");
        //发送消息  fanout中的routingkey没啥作用
        channel.basicPublish("logs","",null,"fanout type message".getBytes());
        RabbitUtils.closeConnectionAndChannel(channel,connection);
    }

4.5.2 开发3个消费者

消费者1:

public class Consumer1 {

    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare("logs","fanout");
        //临时队列
        String queueName = channel.queueDeclare().getQueue();
        //绑定交换机和队列
        channel.queueBind(queueName,"logs","");
        //消费消息
        channel.basicConsume(queueName,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1:"+new String(body));
            }
        });

    }
}

消费者2:

public class Consumer2 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare("logs","fanout");
        //临时队列
        String queueName = channel.queueDeclare().getQueue();
        //绑定交换机和队列
        channel.queueBind(queueName,"logs","");
        //消费消息
        channel.basicConsume(queueName,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2:"+new String(body));
            }
        });

    }
}
public class Consumer3 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare("logs","fanout");
        //临时队列
        String queueName = channel.queueDeclare().getQueue();
        //绑定交换机和队列
        channel.queueBind(queueName,"logs","");
        //消费消息
        channel.basicConsume(queueName,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者3:"+new String(body));
            }
        });

    }
}

先启动3个消费者

启动生产者之后观察3个消费者是否都接收到了消息:

RabbitMQ基础与实操复习

RabbitMQ基础与实操复习

RabbitMQ基础与实操复习

观察是否创建了对应的交换机:

RabbitMQ基础与实操复习

4.6 第四种模型:Routing

  其实Routing和Topics很像,一个是写死了RoutingKey,另一个使用了通配符。

  在Fanout模式中,一条消息,会被所有订阅的队列都消费。但是,在某些场景下,我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

在Direct模型下:

  • 队列与交换机的绑定,不能是任意绑定了,而是要指定一个RoutingKey(路由key)
  • 消息的发送方在 向 Exchange发送消息时,也必须指定消息的 RoutingKey
  • Exchange不再把消息交给每一个绑定的队列,而是根据消息的Routing Key进行判断,只有队列的Routingkey与消息的 Routing key完全一致,才会接收到消息

RabbitMQ基础与实操复习

图解:

  • P:生产者,向Exchange发送消息,发送消息时,会指定一个routing key。
  • X:Exchange(交换机),接收生产者的消息,然后把消息递交给 与routing key完全匹配的队列
  • C1:消费者,其所在队列指定了需要routing key 为 error 的消息
  • C2:消费者,其所在队列指定了需要routing key 为 info、error、warning 的消息

4.6.1 生产者

public class Provider {
    public static final String EXCHANGE_NAME="logs_direct";
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();

        //声明交换机  参数1:交换机名称,参数2:direct 路由模式
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        //发送消息
//        String routingKey="info";
//        String routingKey="error";
        String routingKey="warning";
//        String routingKey="trade";
        channel.basicPublish(EXCHANGE_NAME,routingKey,null,("这是direct模型发布的基于routingKey:["+routingKey+"]").getBytes());

        //关闭资源
        RabbitUtils.closeConnectionAndChannel(channel,connection);
    }
}

4.6.2 消费者1

public class Consumer1 {
    public static final String EXCHANGE_NAME="logs_direct";
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();
        //通道声明交换机以及交换机的类型
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

        //创建一个临时队列
        String queue = channel.queueDeclare().getQueue();
        //基于routingKey去绑定队列和交换机
        channel.queueBind(queue,EXCHANGE_NAME,"error");

        //消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1:"+new String(body));
            }
        });
    }
}

4.6.3 消费者2

public class Consumer2 {
    public static final String EXCHANGE_NAME="logs_direct";
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();
        //通道声明交换机以及交换机的类型
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

        //创建一个临时队列
        String queue = channel.queueDeclare().getQueue();
        //基于routingKey去绑定队列和交换机
        channel.queueBind(queue,EXCHANGE_NAME,"info");
        channel.queueBind(queue,EXCHANGE_NAME,"error");
        channel.queueBind(queue,EXCHANGE_NAME,"warning");

        //消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2:"+new String(body));
            }
        });
    }
}

先启动两个消费者:

RabbitMQ基础与实操复习

启动生产者之后观察消费者是否收到了消息:

RabbitMQ基础与实操复习

RabbitMQ基础与实操复习

  可以看到,消费者1没有收到消息,因为我们生产者的routintKey为warning,而消费者1队列的routingKey是error,消费者2队列的routingKey是warning

  所以只有消费者2可以接收到消息,只有队列的Routingkey与消息的 Routing key完全一致,才会接收到消息

4.7 第五种模型:Topics

  Topic类型的ExchangeDirect相比,都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符!这种模型Routingkey 一般都是由一个或多个单词组成,多个单词之间以”.”分割,例如: item.insert

RabbitMQ基础与实操复习

# 统配符
		* (star) can substitute for exactly one word.    匹配不多不少恰好1个词
		# (hash) can substitute for zero or more words.  匹配一个或多个词
# 如:
	audit.#    匹配audit.irs.corporate或者 audit.irs 等
    audit.*   只能匹配 audit.irs

4.7.1 生产者

public class Provider {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();

        //声明交换机以及交换机类型 topic
        channel.exchangeDeclare("topics", BuiltinExchangeType.TOPIC);

        //发布消息
        String routingKey="user.save";
//        String routingKey="user.save.findAll";
    //    String routingKey="user";
        channel.basicPublish("topics",routingKey,null,("这里是topic动态路由模型,routingKey:["+routingKey+"]").getBytes());

        //关闭资源
        RabbitUtils.closeConnectionAndChannel(channel,connection);
    }
}

4.7.2 消费者1:

public class Cosumer1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();

        //声明交换机以及交换机类型
        channel.exchangeDeclare("topics", BuiltinExchangeType.TOPIC);
        //创建临时队列
        String queue = channel.queueDeclare().getQueue();
        //创建队列和交换机,动态通配符形式 routingKey
        channel.queueBind(queue,"topics","user.*");

        //消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1:"+new String(body));
            }
        });

    }
}

4.7.3 消费者2:

public class Cosumer2 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();

        //声明交换机以及交换机类型
        channel.exchangeDeclare("topics", BuiltinExchangeType.TOPIC);
        //创建临时队列
        String queue = channel.queueDeclare().getQueue();
        //创建队列和交换机,动态通配符形式 routingKey
        channel.queueBind(queue,"topics","user.#");

        //消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2:"+new String(body));
            }
        });

    }
}

我们注意到消费者1的routingKey为user.*,消费者2的routingKey为user.#

启动两个消费者,再启动生产者

RabbitMQ基础与实操复习

RabbitMQ基础与实操复习

此时都收到了消息,是因为两个规则都能匹配到。

我们现在将生产者交换机的routingKey改为user.save.findAll,启动生产者,观察结果:

RabbitMQ基础与实操复习

RabbitMQ基础与实操复习

  是因为消费者2中队列的routingKey为user.#,user后面可以匹配一个或者多个,而消费者1中队列的routingKey为user.*,user后面只能匹配一个词,所以收不到消息。

  就先介绍到这里,后面的RPC暂时不搞了,至于Publisher Confirms看我专栏以前的文章,这几种模式足够应付绝大多数的业务场景了。

5、RabbitMQ与SpringBoot整合

  真正写代码的时候都是与现有框架进行集成,很少用上面那种原生的写法。

5.0 搭建环境

5.0.1 引入依赖

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

5.0.2 配置文件

spring:
  application:
    name: rabbitmq-springboot

  rabbitmq:
    host: 你的ip
    port: 5672
    username: 用户名
    password: 密码
    virtual-host: /ems  # 虚拟主机

5.1 第一种:HelloWorld模型

生产者:

//注入rabbitTemplate
@Autowired
private RabbitTemplate rabbitTemplate;

//hello world
@Test
public void testHelloWorld(){
    rabbitTemplate.convertAndSend("hello","hello world");
}

消费者:

@Component
@RabbitListener(queuesToDeclare = @Queue(value = "hello"))
public class HelloCustomer {

    @RabbitHandler
    public void receive(String message){
        System.out.println("message="+message);
    }
}

由于@RabbitListener注解会一直监听消息,所以这里不用像上面一样分别启动消费者和生产者了。

我们直接启动生产者:

RabbitMQ基础与实操复习

RabbitMQ基础与实操复习

管理界面却是存在hello队列,控制台也看到消息已经被消费者接收。

5.2 第二种:Work Queue

生产者:

//work queues
@Test
public void testWorkQueue(){
    for (int i = 0; i < 10; i++) {
        rabbitTemplate.convertAndSend("work","work模型"+i);
    }
}

消费者:

@Component
public class WorkCustomer {

    //消费者1
    @RabbitListener(queuesToDeclare = @Queue("work"))
    public void receive(String message){
        System.out.println("message1="+message);
    }

    //消费者2
    @RabbitListener(queuesToDeclare = @Queue("work"))
    public void receive1(String message){
        System.out.println("message2="+message);
    }
}

这里我建立两个消费者来测试轮询分发模式

RabbitMQ基础与实操复习

RabbitMQ基础与实操复习

管理界面中看到了绑定的work队列,控制台也看到了消息已经被两个消费者接收。

5.3 第三种:Fanout

生产者:

//fanout 广播
@Test
public void testFanout(){
    //这种模式的routingKey没啥作用
    rabbitTemplate.convertAndSend("logs","","Fanout的模型发送的消息");
}

定义一个交换机:logs

消费者:

@Component
public class FanoutCustomer {

    @RabbitListener(bindings = {
           @QueueBinding(
                   value = @Queue,//不写value代表临时队列
                   exchange =@Exchange(value = "logs",type ="fanout")   //绑定的叫喊及
           )
    })
    public void receive1(String message){
        System.out.println("message1= "+message);
    }

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,//不写value代表临时队列
                    exchange =@Exchange(value = "logs",type ="fanout")   //绑定的交换机
            )
    })
    public void receive2(String message){
        System.out.println("message2= "+message);
    }
}

  这里两个临时队列都与logs交换机进行绑定,所以我们生产者将消息发送到logs交换机上面之后,两个消费者都能接收到消息。

RabbitMQ基础与实操复习

RabbitMQ基础与实操复习

5.4 第四种:Routing

生产者:

 //routing 路由模式
    @Test
    public void testRoute(){
//        rabbitTemplate.convertAndSend("directs","info","发送info的key的路由信息");
        rabbitTemplate.convertAndSend("directs","error","发送info的key的路由信息");
    }

消费者:

@Component
public class RouteCustomer {

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue, //创建临时队列
                    exchange = @Exchange(value = "directs",type = "direct"),//指定交换机
                    key = {"info","error","warn"}
            )
    })
    public void receive1(String message){
        System.out.println("message1= "+message);
    }

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue, //创建临时队列
                    exchange = @Exchange(value = "directs",type = "direct"),//指定交换机
                    key = {"error"}
            )
    })
    public void receive2(String message){
        System.out.println("message2= "+message);
    }
}

当routingKey=error的时候,两个消费者都可以接收到:

RabbitMQ基础与实操复习

我们现在将routingKey改为info,再次发送:

rabbitTemplate.convertAndSend("directs","info","发送info的key的路由信息");

RabbitMQ基础与实操复习

  可以看到,只有消费者1接收到了消息,因为只有消费者1的队列和交换机进行绑定的routingKey"info","error","warn",包含了info,而消费者2中队列和交换机绑定的routingKeyerror,所以消费者2接收不到这条消息。

5.4 第五种:Topics

  也叫动态路由模型,就是在第四种模型的基础之上加了通配符而已。

生产者:

  //topic 动态路由  订阅模式
    @Test
    public void testTopic(){
        rabbitTemplate.convertAndSend("topics","user.save","user.save 路由消息");
//        rabbitTemplate.convertAndSend("topics","order","user.save 路由消息");
//        rabbitTemplate.convertAndSend("topics","product.save.add","product.save.add 路由消息");
    }

消费者:

@Component
public class TopicCustomer {

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,
                    exchange = @Exchange(value = "topics",type = "topic"),
                    key = {"user.save","user.*"}
            )
    })
    public void reveive1(String message){
        System.out.println("message1 = "+message);
    }

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,
                    exchange = @Exchange(value = "topics",type = "topic"),
                    key = {"order.#","product.#","user.*"}
            )
    })
    public void reveive2(String message){
        System.out.println("message2 = "+message);
    }
}

  此时,生产者中的定义的routingKeyuser.save,而消费者1有user.save和user.*,消费者2有:user.*,所以两个都能接收到消息:

RabbitMQ基础与实操复习

RabbitMQ基础与实操复习

控制台看到两个消费者都输出了消息,管理界面中也看到了新建的交换机。

现在修改生产者消息的routingKey如下:

rabbitTemplate.convertAndSend("topics","order","user.save 路由消息");

RabbitMQ基础与实操复习

  可以看到,只有消费者2接收到了消息,这是因为消费者中的routingKey包含"order.#",#代表有一个或者多个单词,所以匹配到。

将生产者代码修改如下:

rabbitTemplate.convertAndSend("topics","product.save.add","product.save.add 路由消息");

RabbitMQ基础与实操复习

消费者2中有routingKeyproduct.#,所以能够接收到。文章来源地址https://www.toymoban.com/news/detail-498003.html

到了这里,关于RabbitMQ基础与实操复习的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Redis数据库持久化---RDB(Redis DataBase)概念与实操

    一、概念 1redis提供了两种持久化的方式,分别是RDB(Redis DataBase)和AOF(Append Only File)。 RDB(Redis DataBase):是在不同的时间点,将redis存储的数据生成快照并存储到磁盘等介质上; 特点: 1.周期性 2.不影响数据写入  #RDB会启动子进程,备份所有数据。当前进程,继续提供数

    2024年02月10日
    浏览(35)
  • R数据分析:多项式回归与响应面分析的理解与实操

    今天给大家分享一个新的统计方法,叫做响应面分析,响应面分析是用来探究变量一致性假设的(Congruence hypotheses)。本身是一个工程学方法,目前在组织行为学,管理,市场营销等等领域中使用越来越多。 Congruence hypotheses state that the agreement (i.e., congruence) between two construc

    2024年02月07日
    浏览(47)
  • LLaMA 简介:一个具有 650 亿参数的基础大型语言模型 官网介绍

    2023 年 2 月 24 日 更新: 我们刚刚推出了Llama 2 - 有关最新信息的更多信息,请参阅我们关于 Llama 2 的博客文章。 作为 Meta 致力于开放科学的一部分,今天我们公开发布 LLaMA(大型语言模型 Meta AI),这是一种最先进的基础大型语言模型,旨在帮助研究人员推进人工智能这一子

    2024年02月10日
    浏览(38)
  • (三)RabbitMQ七种模式介绍与代码演示

    Lison dreamlison@163.com , v1.0.0 , 2023.06.22 交换机概念 交换机可以理解成具有路由表的路由程序,仅此而已。每个消息都有一个称为路由键(routing key)的属性,就是一个简单的字符串。 最新版本的RabbitMQ有四种交换机类型,分别是Direct exchange、Fanout exchange、Topic exchange、Headers exc

    2024年02月15日
    浏览(28)
  • RabbitMQ的基本概念和七种队列模式

    1. 生产者/消费者 生产者(Producer) 消息的创建者。 负责创建和推送数据到消息服务器。 消费者(Consumer) 消息的接收方。 负责接收消息和处理数据。 2. 消息队列(Queue) 消息队列是RabbitMQ的内部对象,用于存储生产者的消息直到发送给消费者,它是消费者接收消息的地方。 消息队

    2024年02月13日
    浏览(30)
  • 大数据技术期末复习重点,不挂科看这里~

    本篇文章是根据老师画的重点,作为期末大数据复习资料,持续更新中~ 目录 一、大数据概念 1.什么是大数据,大数据的意义是什么?大数据的特征是什么? 2.大数据的技术支撑是什么? 3.大数据的处理方法有哪些? 二、数据采集与预处理 1.Kafka数据采集 2.数据预处理原理

    2024年02月11日
    浏览(42)
  • 【零基础学机器学习 2】 机器学习的实操步骤-以及在Python中实现机器学习模型

    👨‍💻 作者简介:程序员半夏 , 一名全栈程序员,擅长使用各种编程语言和框架,如JavaScript、React、Node.js、Java、Python、Django、MySQL等.专注于大前端与后端的硬核干货分享,同时是一个随缘更新的UP主. 你可以在各个平台找到我! 🏆 本文收录于专栏: 零基础学机器学习 🔥 专栏

    2024年02月05日
    浏览(40)
  • rabbitmq-常见七种消息队列-控制台界面管理-python-实现简单访问

    1.1.生产者和消费者 生产者(Producer) 消息的创建者。 负责创建和推送数据到消息服务器。 消费者(Consumer) 消息的接收方。 负责接收消息和处理数据。 1.2.消息队列(Queue) 消息队列是RabbitMQ的内部对象,用于存储生产者的消息直到发送给消费者,它是消费者接收消息的地方。 消息

    2024年02月03日
    浏览(65)
  • 人工智能学习与实训笔记(九):Langchain + 百度大模型实战案例

    人工智能专栏文章汇总:人工智能学习专栏文章汇总-CSDN博客 本篇目录 1. LLMChain 2. Sequential Chains SimpleSequentialChain SequentialChain 3. Router Chain 4. Documents Chain 5. Document Loaders(文档加载器) 6. Text Splitter(文本分割器) 7. Vectorstores(向量存储器) 8. Retriever(检索器) 9. Agent(代

    2024年04月13日
    浏览(46)
  • 人工智能学习与实训笔记(一):零基础理解神经网络

    人工智能专栏文章汇总:人工智能学习专栏文章汇总-CSDN博客 本篇目录 一、什么是神经网络模型 二、机器学习的类型 2.1 监督学习 2.2 无监督学习 2.3 半监督学习 2.4 强化学习 三、网络模型结构基础 3.1 单层网络 ​编辑 3.2 多层网络 3.3 非线性多层网络  四、 神经网络解决回

    2024年02月20日
    浏览(41)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包