栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 前沿技术 > 大数据 > 大数据系统

RabbitMQ的创建及使用

RabbitMQ的创建及使用

RabbitMQ 是由 erlang 开发的 AMQP 的开源实现

核心概念

Message:消息

Publisher:消息生产者

Exchange:交换器 (Exchange 有 4 种类型:direct(默认)、fanout、topic 和 headers)

Queue:消息队列

Binding:绑定 (Exchange 和 Queue 的绑定可以是多对多的关联)

Connection:网络连接 (比如一个 TCP 连接)

Channel:信道

Consumer:消息的消费者

Virtual Host:虚拟主机

Broker:消息队列服务器实体

RabbitMQ的创建

开启mq,通过本地链接访问 localhost:15672 ,默认账号密码为:guest

创建交换器

 

创建队列

 

绑定

 

RabbitMQ 使用 

依赖


   org.springframework.boot
   spring-boot-starter-amqp

YML

spring:
  rabbitmq:
    #地址 
    host: 127.0.0.1
    #端口号
    port: 5672
    #账号 密码
    username: guest
    password: guest
    lister:
      #设置监听容器,若不设置,将默认为 SimpleRabbitListenerContainerFactory,且下面的 direct 配置不生效
      #如果不设置 type,要将 direct 改成 simple 才生效
      type: direct
      direct:
        #设置监听为手动答应模式
        acknowledge-mode: manual

RabbitTemplate  (给 RabbitMQ 发送和接收消息)

// 注入
@Autowired 
RabbitTrmplate rabbitTrmplate 
// 点对点(单播)
  //Message 需要自己构造一个,定义消息体内容和消息头
  rabbitTemlpate.send(exchange,routeKey,message);
  
  // object 默认当成消息体  只需要传入要发送的对象,自动序列化发送给 rabbitmq
  // 发送消息
  rabbitTemlpate.convertAndSend(exchange.direct,routeKey,object);
  
  //接收消息(此处的 queueName 是发送消息时路由键多对应的队列)
  rabbitTemlpate.receiveAndConvert(queueName);

// 广播(路由键可以不写,会发送到每个队列中)
  // 发送消息
  rabbitTemlpate.convertAndSend(exchange.fanout,"",object);
  
其他模式只需要把 rabbitMQ 创建好直接使用



// 监听  需要在主类上开启基于注解的 Rabbit 模式   @EnableRabbit
  @RabbitListener(queues = "") // queues 为数组形式,队列可以为多个
  public void receive1(){}
  
  // 获取消息头
  @RabbitListener(queues = "")
  public void receive2(Message message){
    message.getBoby(); // 消息内容
    message.getMessageProperties(); // 消息头信息
  }


自定义序列化格式
@Configuration
public class AmqpConfig {
    // 自定义 json 格式发送消息
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

AmqpAdmin (RabbitMQ 系统管理功能组件(创建和删除 Queue、Exchange、Binding))

// 注入
@Autowired 
AmqpAdmin amqpAdmin

//新建交换器
DirectExchange directExchange = new DirectExchange("name");
//创建交换器
amqpAdmin.declareExchange(directExchange);
//新建队列
Queue queue = new Queue("name",durable); // durable boolean 类型
//创建队列
amqpAdmin.declareQueue(queue);
//新建绑定规则
Binding binding = 
new Binding("destination", Binding.DestinationType.QUEUE,"exchange","routingKey",null);
// 绑定
amqpAdmin.declareBinding(binding);

// 删除交换器
amqpAdmin.deleteExchange();
// 删除队列
amqpAdmin.deleteQueue();

转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/743349.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号