springboot（五）rabbitmq的使用

转x33g5p2x  于2021-12-17 转载在 其他  
字(8.3k)|赞(0)|评价(0)|浏览(304)

rabbitmq 介绍

       rabbitmq是实现 AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。 rabbitmq主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。

AMQP，即 Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP 的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

rabbitmq是一个开源的 AMQP 实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP 等，支持 AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

交换机的四种类型：

      交换机的功能主要是接收消息并且转发到绑定的队列，交换机不存储消息，在启用ack模式后，交换机找不到队列会返回错误。交换机有四种类型：Direct, topic, Headers and Fanout

1.Direct Exchange

      Direct Exchange 是 RabbitMQ 默认的交换机模式，也是最简单的模式，根据key全文匹配去寻找队列。

2.Topic Exchange

*       *Topic Exchange 转发消息主要是根据通配符。 在这种交换机下，队列和交换机的绑定会定义一种路由模式，那么，通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息。

在这种交换机模式下：

  • 路由键必须是一串字符，用句号（.） 隔开，比如说 agreements.us，或者 agreements.eu.stockholm 等。
  • 路由模式必须包含一个 星号（*），主要用于匹配路由键指定位置的一个单词，比如说，一个路由模式是这样子：agreements..b.*，那么就只能匹配路由键是这样子的：第一个单词是 agreements，第四个单词是 b。 井号（#）就表示相当于一个或者多个单词，例如一个匹配模式是 agreements.eu.berlin.#，那么，以agreements.eu.berlin 开头的路由键都是可以的。

topic 和 direct 类似, 只是匹配上支持了”模式”, 在”点分”的 routing_key 形式中, 可以使用两个通配符:

  • *表示一个词.
  • #表示零个或多个词.

3.Fanout Exchange

     Fanout Exchange 消息广播的模式，不管路由键或者是路由模式，会把消息发给绑定给它的全部队列，如果配置了 routing_key 会被忽略。

4.Headers Exchange

     Headers Exchange 也是根据规则匹配, 相较于 direct 和 topic 固定地使用 routing_key , headers 则是一个自定义匹配规则的类型. 在队列与交换器绑定时, 会设定一组键值对规则, 消息中也包括一组键值对( headers 属性), 当这些键值对有一对, 或全部匹配时, 消息被投送到对应队列.

安装rabbitmq

     使用docker镜像安装rabbitmq，简单快捷。教程地址：

     https://blog.csdn.net/xu12387/article/details/85049486

springboot集成rabbitmq

     新建项目：springboot-rabbitmq ，打开pom.xml文件加入相关依赖

<dependencies>
		<!--测试-->
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
		<!--rabbitmq-->
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-amqp</artifactId>
		</dependency>
    </dependencies>

      配置application.properties，内容如下：

#rabbitmq主机
spring.rabbitmq.host=10.24.247.23
#端口
spring.rabbitmq.port=5672
#用户名
spring.rabbitmq.username=admin
#密码
spring.rabbitmq.password=123456

1.交换机的四种模式实现：Direct Exchange模式（默认模式）：

         配置文件RabbitMQConfig .java，内容如下：

@Configuration
public class RabbitMQConfig {
    public static final String QUEUE = "queue";

    /**
     * Directm模式
     */
    @Bean
    public Queue queue() {
        //true代表是否持久化
        return new Queue(QUEUE, true);
    }

}

       发送者Sender.java，内容如下：

@Component
public class Sender {
    private static Logger log = LoggerFactory.getLogger(Sender.class);

    @Autowired
    private AmqpTemplate amqpTemplate;
    
    public void send(String msg){
        log.info("send message:"+msg);
        amqpTemplate.convertAndSend(RabbitMQConfig.QUEUE,msg);
    }
}

        接收者Recevier.java，内容如下：

@Component
public class Receiver {
    private static Logger log = LoggerFactory.getLogger(Receiver.class);
    
    @RabbitListener(queues = RabbitMQConfig.QUEUE)
    public void receiver(String message) {
        log.info("recevier message:"+message);
    }
}

        编写测试方法，内容如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringbootRabbitmqApplicationTests {

    @Autowired
    private Sender sender;
    @Test
    public void send() {
        sender.send("你好，我是老猫");
    }
}

       可以清楚的看到，发送者只要发送消息，接收者就能立马收到。特别注意，他们的queue(队列)都是同一个！

2.交换机的四种模式实现：Direct Exchange模式：

        配置文件RabbitMQConfig .java，内容如下：

@Configuration
public class RabbitMQConfig {
    public static final String TOPIC_QUEUE1 = "topic.queue1";
    public static final String TOPIC_QUEUE2 = "topic.queue2";
    public static final String TOPIC_EXCHANGE = "topicExchange";
    /**
     * Topic模式
     */
    @Bean
    public Queue topicQueue1() {
        return new Queue(TOPIC_QUEUE1, true);
    }

    @Bean
    public Queue topicQueue2() {
        return new Queue(TOPIC_QUEUE2, true);
    }

    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }
    @Bean
    public Binding topicBinding1(){
        return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("topic.key1");
    }
    @Bean
    public Binding topicBinding2(){
        return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("topic.#");
    }

}

       发送者Sender.java，内容如下：

@Component
public class Sender {
    private static Logger log = LoggerFactory.getLogger(Sender.class);

    @Autowired
    private AmqpTemplate amqpTemplate;

    public void sendTopic(String msg) {
        log.info("send topic message:" + msg);
        amqpTemplate.convertAndSend(RabbitMQConfig.TOPIC_EXCHANGE, "topic.key1", msg + 1);
        amqpTemplate.convertAndSend(RabbitMQConfig.TOPIC_EXCHANGE, "topic.key2", msg + 2);
    }
}

        接收者Recevier.java，内容如下：

@Component
public class Receiver {
    private static Logger log = LoggerFactory.getLogger(Receiver.class);

    @RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE1)
    public void receiverTopic1(String message) {
        log.info("receiver topic queue1 message:"+message);
    }

    @RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE2)
    public void receiverTopic2(String message) {
        log.info("receiver topic queue2 message:"+message);
    }
}

        编写测试方法，内容如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringbootRabbitmqApplicationTests {

    @Autowired
    private Sender sender;
    @Test
    public void sendTopic() {
        sender.sendTopic("你好，我是老猫");
    }
}

其中topic.#中的#代表匹配任意单词，所以我们的queue2就就收到了两次，queue1因为binding的是固定名称所以只接收了一次

3.交换机的四种模式实现：Fanout Exchange模式：

        配置文件RabbitMQConfig .java，内容如下：

@Configuration
public class RabbitMQConfig {
    public static final String FANOUT_QUEUE1 = "fanout.queue1";
    public static final String FANOUT_QUEUE2 = "fanout.queue2";
    public static final String FANOUT_EXCHANGE = "fanoutExchange";

    /**
     * Fanout模式
     */
    @Bean
    public Queue fanoutQueue1() {
        return new Queue(FANOUT_QUEUE1, true);
    }

    @Bean
    public Queue fanoutQueue2() {
        return new Queue(FANOUT_QUEUE2, true);
    }

    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(FANOUT_EXCHANGE);
    }

    @Bean
    public Binding fanoutBinding1() {
        return BindingBuilder.bind(fanoutQueue1()).to(fanoutExchange());
    }

    @Bean
    public Binding fanoutBinding2() {
        return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
    }

}

       发送者Sender.java，内容如下：

@Component
public class Sender {
    private static Logger log = LoggerFactory.getLogger(Sender.class);
    
    @Autowired
    private AmqpTemplate amqpTemplate;

    public void sendFanout(String msg){
        log.info("send fanout message:"+msg);
        amqpTemplate.convertAndSend(RabbitMQConfig.FANOUT_EXCHANGE,"",msg);
    }

}

        接收者Recevier.java，内容如下：

@Component
public class Receiver {
    private static Logger log = LoggerFactory.getLogger(Receiver.class);

    @RabbitListener(queues = RabbitMQConfig.FANOUT_QUEUE1)
    public void receiverFanout1(String message) {
        log.info("receiver fanout queue1 message:"+message);
    }

    @RabbitListener(queues = RabbitMQConfig.FANOUT_QUEUE2)
    public void receiverFanout2(String message) {
        log.info("receiver fanout queue2 message:"+message);
    }
}

        编写测试方法，内容如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringbootRabbitmqApplicationTests {

    @Autowired
    private Sender sender;

    @Test
    public void sendFanout() {
        sender.sendFanout("你好，我是老猫");
    }
}

4.交换机的四种模式实现：Fanout Exchange模式：

        配置文件RabbitMQConfig .java，内容如下：

@Configuration
public class RabbitMQConfig {
    public static final String HEADERS_QUEUE = "headers.queue";
    public static final String HEADERS_EXCHANGE = "headersExchange";

    /**
     * Headers模式
     */
    @Bean
    public HeadersExchange headersExchange() {
        return new HeadersExchange(HEADER_EXCHANGE);
    }
    @Bean
    public Queue headersQueue() {
        return new Queue(HEADERS_QUEUE, true);
    }
    @Bean
    public Binding headersBinding(){
        Map<String,Object> map=new HashMap<String,Object>();
        map.put("headers1","value1");
        map.put("headers2","value2");
        return BindingBuilder.bind(headersQueue()).to(headersExchange()).whereAll(map).match();
    }

}

       发送者Sender.java，内容如下：

@Component
public class Sender {
    private static Logger log = LoggerFactory.getLogger(Sender.class);

    @Autowired
    private AmqpTemplate amqpTemplate;

    public void sendHeaders(String msg){
        log.info("send headers message:"+msg);
        MessageProperties properties=new MessageProperties();
        properties.setHeader("headers1","value1");
        properties.setHeader("headers2","value2");
        Message message=new Message(msg.getBytes(),properties);
        amqpTemplate.convertAndSend(RabbitMQConfig.HEADERS_EXCHANGE,"", message);
    }

}

        接收者Recevier.java，内容如下：

@Component
public class Receiver {
    private static Logger log = LoggerFactory.getLogger(Receiver.class);

    @RabbitListener(queues = MQConfig.HEADERS_QUEUE)
    public void receiverHeaderS(byte[] message) {
        log.info("receiver headers message:"+new String(message));
    }
}

        编写测试方法，内容如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringbootRabbitmqApplicationTests {

    @Autowired
    private Sender sender;

    @Test
    public void sendHeaders(){
        sender.sendHeaders("你好，我是老猫");
    }
}

源码地址：https://gitee.com/xu0123/springboot2

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/xu12387/article/details/88637572

内容来源于网络，如有侵权，请联系作者删除！

Javaspringbootrabbitmq

相关文章

    查看更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com