RabbitMQ 消息队列使用_代码007(未授权)

本文介绍: 话题交换机的routingkey 必须是多个单词的列表,并以。1.声明交换机, 队列,并绑定,添加routingkey。Direct交换机会将消息按照路由规则发送到指定的队列。Fanout交换将将消息发送到每一个绑定到它的队列中。3.发消息到fanoutexchange。RabbitMQ默认使用JDK自带序列化。发送对象类型过去, 查看序列化后的值。基于 Spring Amqp。2.消费者监听这两个队列。时效性强,立即得到结果。:代表0个或者多个单词。

同步调用优点:

时效性强,立即得到结果

缺点:

异步调用优点:

基于 Spring Amqp

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

spring:
  rabbitmq:
    port: 5672
    host: localhost
    virtual-host: /
    username: guest
    password: guest

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

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void helloWorldModel(){
        String queueName="zbq.queue1";
        String message="hello, spring amqp";
        rabbitTemplate.convertAndSend(queueName,message);
    }
}

@Component
@Slf4j
public class SpringRabbitListener {
    @RabbitListener(queues = "zbq.queue1")
    public void listenSimpleQueue(String msg){
        log.info("收到消息: "+msg);
    }
}

@Test
    public void workQueueModel(){
        String queueName="zbq.work.queue";
        String msg="hello, amqp ";
        for(int i=0;i<1000;i++){
            rabbitTemplate.convertAndSend(queueName,msg+i);
        }
    }

@RabbitListener(queues = "zbq.work.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
    log.info("消费者1号接收到消息: "+msg);
    Thread.sleep(20);
}

@RabbitListener(queues = "zbq.work.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
    log.info("消费者2号接收到消息: "+msg);
    Thread.sleep(100);
}

@Configuration
public class FanoutConfig {
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("zbq.fanout");
    }

    @Bean
    public Queue fanoutQueue1(){
        return new Queue("zbq.fanout.queue1");
    }

    @Bean
    public Binding bindingQueue1(Queue fanoutQueue1,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    @Bean
    public Queue fanoutQueue2(){
        return new Queue("zbq.fanout.queue2");
    }

    @Bean
    public Binding bindingQueue2(Queue fanoutQueue2,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

@RabbitListener(queues = "zbq.fanout.queue1")
public void listenFanoutQueue1(String msg){
    log.info("消费者1收到Fanout消息: "+msg);
}

@RabbitListener(queues = "zbq.fanout.queue2")
public void listenFanoutQueue2(String msg){
    log.info("消费者2收到Fanout消息: "+msg);
}

@Test
public void fanoutModel(){
    String exchangeName="zbq.fanout";
    String msg="hello, fanout ";
    for(int i=0;i<10;i++){
        rabbitTemplate.convertAndSend(exchangeName,"",msg+i);
    }
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "zbq.direct.queue1"),
    exchange = @Exchange(name="zbq.direct",type = ExchangeTypes.DIRECT),
    key = {"girlfriend","family"}
))
public void listenDirectQueue1(String msg){
    log.info("消费者1收到Direct消息: "+msg);
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "zbq.direct.queue2"),
    exchange = @Exchange(name="zbq.direct",type = ExchangeTypes.DIRECT),
    key = {"friend","family"}
))
public void listenDirectQueue2(String msg){
    log.info("消费者2收到Direct消息: "+msg);
}

@Test
public void directModel(){
    String exchangeName="zbq.direct";
    String msg="晚上回去吃饭 ";
    for (int i = 0; i < 10; i++) {
        rabbitTemplate.convertAndSend(exchangeName,"family",msg+i);
    }
}

@Test
public void directModel2(){
    String exchangeName="zbq.direct";
    String msg="hello, direct";
    for (int i = 0; i < 10; i++) {
        rabbitTemplate.convertAndSend(exchangeName,"girlfriend",msg+i);
    }
}

话题交换机的routingkey 必须是多个单词的列表,并以.分隔

可以使用通配符#和*

#:代表0个或者多个单词

*:代表1个单词

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "zbq.topic.queue1"),
    exchange = @Exchange(name="zbq.topic",type = ExchangeTypes.TOPIC),
    key = "China.#"
))
public void listenTopicQueue1(String msg){
    log.info("消费者1收到Topic消息: "+msg);
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "zbq.topic.queue2"),
    exchange = @Exchange(name="zbq.topic",type = ExchangeTypes.TOPIC),
    key = "#.weather"
))
public void listenTopicQueue2(String msg){
    log.info("消费者2收到Topic消息: "+msg);
}

@Test
public void topicModel(){
    String exchangeName="zbq.topic";
    String msg="首都北京,今日气温10摄氏度";
    rabbitTemplate.convertAndSend(exchangeName,"China.weather",msg);
}

@Test
public void test(){
    Map<String,Object> map=new HashMap<>();
    map.put("hair","long");
    map.put("eyes","big");
    rabbitTemplate.convertAndSend("zbq.queue1",map);
}

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

@Bean
public MessageConverter customMC(){
    return new Jackson2JsonMessageConverter();
}

显示所有内容

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。

rabbitmq 序列化队列

文章目录

MQ

同步调用和异步调用

RabbitMQ

基础概念

模型

简单队列 (Hello-World)

工作队列 (Work Queue)

Pub/Sub (Fanout exchange)

Direct Exchange

Topic Exchange

序列化方式

发表回复取消回复