no13-高并发-rabbitmq消息队列

飞一样的编程

擅长邻域：Java,MySQL,Linux,nginx,springboot,mongodb,微信小程序,vue

分类： springboot 专栏： springboot学习标签：消息队列

2023-04-06 11:57:31 714浏览

消息队列学习

简介

RabbitMQ是个由erlang开发的AMQP(Advanved Message Queue Protocol)的开源实现。翻译过来就是高级消息队列协议

应用场景

异步处理

案例：用户注册成功后，给该用户发送邮件和短信提醒其注册成功

采用消息队列后可以增强用户体验，让用户更快地得到响应。

应用解耦

当用户下单后要对库存系统进行一系列增删改查操作。传统方式的话耦合性太高。

或者是两个系统压根不是同一种语言开发的，比如一个是java开发的，一个是Python开发的，也可以用消息队列的方式做到解耦。

流量削峰

案例：10万人秒杀1万个奖品。

这里就可以创建一个1万个座位的消息队列，开始活动时，当这一万个座位已经抢完后，剩下的9万人再点击秒杀按钮的时候立马返回秒杀失败谢谢参与的提醒，而占到座位的那1万个客户再由队列慢慢处理消化。

核心概念

消息message

由消息头和消息体组成。

消息体是不透明的，而消息头是由一系列的可选属性组成的。

这些属性routing-key (路由键)、priority (相对于其他消息的优先权)、delivery-mode (指出

该消息可能需要持久性存储)等。

消息的生产者publisher

也是一个向交换器发布消息的客户端应用程序。

交换器exchange

用来接收生产者发送的消息并将这些消息路由给到服务器中队列。

Exchange有4种类型: direct(默认)， fanout, topic,和headers,不同类型的Exchange转发消息的策略有

消息队列queue

用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。

一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。

绑定binding

用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。

Exchange和Queue的绑定可以是多对多的关系。

网络连接connection

比如一个TCP连接。

信道 channel

多路复用连接中的一条独立的双向数据流通道，信道是建立在真实的TCP连接内的虚拟连接。AMQP命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这都是通过信道完成，因为对于操作系统来说建立和销毁TCP都是非常昂贵的开销，所了信道的概念，以复用一条TCP连接。

消费者consumer

表示一个从消息队列中取得消息的客户端应用程序。

虚拟主机virtual host

虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个vhost本质上就是一个mini版的RabbitMQ服务器，拥有自己的队列、交换器、绑定和权限机制。vhost是AMQP概念的基础，必须在连接时指定，RabbitMQ默认的vhost是/。

服务器实体broker

表示消息队列服务器实体

运行机制

核心就是交换机和绑定规则

2.两种通信机制

点对点式

消息发送者发送消息，消息代理将其放入一个队列中，消息接收者从队列中获取消息内容,

消息读取后被移出队列。消息只有唯一的发送者和接受者，但并不是说只能有一个接收者

发布订阅式

发送者(发布者)发送消息到主题，多个接收者(订阅者)监听(订阅)这个主题，那么就会在消息到达时同时收到消息（比如公告之类的）

java使用流程

1.pom

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

2.配置

spring.rabbitmq.host=192.168.56.15

3.创建交换机-消息队列-绑定关系

方式一：配置类的方式

//单播的形式
@Configuration
public class DirectConfig {

  @Bean
  public DirectExchange directExchange() {

      //durable是否实例化持久化，autodelete是否自动删除
    DirectExchange directExchange = new DirectExchange("email_direct");//durable是否实例化，autodelete是否自动删除

    return directExchange;
  }

  @Bean
  public Queue emailQueue() {
    Queue queue = new Queue("email_queue");
    return queue;
  }

  @Bean
  Binding bindingEmailQueue(){
    return BindingBuilder.bind(emailQueue()).to(directExchange()).with("email_direct_queue");
  }

}

方式二：AmqpAdmin

RabbitMQ系统管理功能组件;创建和删除queue exchange binding

    @Autowired
    AmqpAdmin amqpAdmin;

    @Test
    void  testamqp(){

        //创建一个Direct的交换器
        //amqpAdmin.declareExchange(new DirectExchange("amqp.exchange"));
        //创建一个永久的队列
        /*amqpAdmin.declareQueue(new Queue("amqp.queue",true));*/
        //创建一个绑定规则
        amqpAdmin.declareBinding(new Binding("amqp.queue", Binding.DestinationType.QUEUE,"amqp.exchange","amqp.hh",null));

    }

4.消息发送者

    @Resource
    RabbitTemplate rabbitTemplate;
    //消息提供者
    @Test
    void testSend() {
        rabbitTemplate.convertAndSend("email_direct","email_direct_queue","141134487@qq.com");
    }

这里注意下：如果发送的是对象或者list之类的这种数据的话，要考虑下序列化的问题，默认采用的是jvm的序列化方式，修改成json序列化方式

@Configuration
public class MyAMQPConfig {
    @Bean
    public  MessageConverter messageConverter(){
       return new Jackson2JsonMessageConverter();
    }
}

5.消息消费

@Component
public class EmailConsumer {

   //模拟多个消费者
    @RabbitListener(queues = "email_queue")
    public void  show(String message){
        System.out.println("客户1接收到的邮件消息："+message);
    }
    @RabbitListener(queues = "email_queue")
    public void  show2(String message){
        System.out.println("客户2接收到的邮件消息："+message);
    }
}

如果是有多个消费者同时监听一个队列的话，默认是采取轮询的机制。比如上面的代码就是第一次客户1接收消息第二次的时候就是客户2接收消息了。

广播交换机

创建交换机、队列、绑定关系

@Configuration
public class FanoutConfig {

  @Bean
  public FanoutExchange newsExchange() {//公司新闻交换机
    //durable是否实例化持久化，autodelete是否自动删除
    FanoutExchange fanoutExchange = new FanoutExchange("news_fanout");

    return fanoutExchange;
  }

  @Bean
  public Queue itQueue() {//技术部新闻队列
    Queue queue = new Queue("it_news_queue");
    return queue;
  }
  @Bean
  public Queue marketQueue() {//市场部新闻队列
    Queue queue = new Queue("market_news_queue");
    return queue;
  }

  @Bean
  Binding bindingQueue1(){//绑定关系
    return BindingBuilder.bind(itQueue()).to(newsExchange());
  }
  @Bean
  Binding bindingQueue2(){//绑定关系
    return BindingBuilder.bind(marketQueue()).to(newsExchange());
  }

}

发送消息

  //测试广播
    @Test
    void testFanout(){
        rabbitTemplate.convertAndSend("news_fanout", null,"杰凡it放假10天");
    }

接收消息

@Component
public class NewsConsumer {

    @RabbitListener(queues = "it_news_queue")
    public void  show(String message){

        System.out.println("技术部接收到的消息："+message);
    }
    @RabbitListener(queues = "market_news_queue")
    public void  getmes(String message){

        System.out.println("市场部接收到的消息："+message);
    }
}

主题交换机

https://blog.csdn.net/weixin_44741023/article/details/128520910

创建交换机、队列、绑定关系

@Component
@Configuration
public class TopicConfig {
    @Bean
    public TopicExchange topicExchange() {
        //durable是否实例化持久化，autodelete是否自动删除
        TopicExchange directExchange = new TopicExchange("exchange.topic");

        return directExchange;
    }

    //湖南的消息队列
    @Bean
    public Queue hunanQueue() {
        Queue queue = new Queue("hunan.queue");
        return queue;
    }

    //湖北的消息队列
    @Bean
    public Queue hubeiQueue() {
        Queue queue = new Queue("hubei.queue");
        return queue;
    }

    //新闻的消息队列
    @Bean
    public Queue newsQueue() {
        Queue queue = new Queue("news.queue");
        return queue;
    }

    //天气的消息队列
    @Bean
    public Queue weatherQueue() {
        Queue queue = new Queue("weather.queue");
        return queue;
    }
    @Bean
    Binding bindingChinaQueue(){
        //#：代表0个或多个关键词
        //*：代表1个单词
        return BindingBuilder.bind(hunanQueue()).to(topicExchange()).with("hunan.#");
    }
    @Bean
    Binding bindingJapanQueue(){
        return BindingBuilder.bind(hubeiQueue()).to(topicExchange()).with("hubei.#");
    }
    @Bean
    Binding bindingNewsQueue(){
//        return BindingBuilder.bind(newsQueue()).to(topicExchange()).with("#.news");
        return BindingBuilder.bind(newsQueue()).to(topicExchange()).with("*.news");
    }
    @Bean
    Binding bindingWeatherQueue(){
        return BindingBuilder.bind(weatherQueue()).to(topicExchange()).with("#.weather");
    }



}

发送消息

    //测试主题
    @Test
    void testTopic(){
        rabbitTemplate.convertAndSend("exchange.topic", "hunan.yueyang.news","湖南岳阳的新闻");
//        rabbitTemplate.convertAndSend("exchange.topic", "hubei.news","湖北新闻");
//        rabbitTemplate.convertAndSend("exchange.topic", "hunan.weather","湖南天气");
//        rabbitTemplate.convertAndSend("exchange.topic", "hubei.weather","湖北天气");
    }

接收消息

@Component
public class TopicConsumer {
    @RabbitListener(queues = "hunan.queue")
    public void  getHunan(String message){

        System.out.println("监听湖南的信息："+message);
    }
    @RabbitListener(queues = "hubei.queue")
    public void  getHubei(String message){

        System.out.println("监听湖北的消息："+message);
    }
    @RabbitListener(queues = "news.queue")
    public void  getNews(String message){

        System.out.println("监听新闻消息："+message);
    }
    @RabbitListener(queues = "weather.queue")
    public void  get(String message){

        System.out.println("监听天气消息："+message);
    }
}