栏目分类:
子分类:
返回
终身学习网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
终身学习网 > IT > 软件开发 > 后端开发 > Java

SpringBoot集成RabbitMq

Java 更新时间:发布时间: 百科书网 趣学号
一:导入依赖

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

二:配置交换机及队列

package com.zy.rabbitMq.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class DelayConfig {
    //延迟交换机
    public static final String delayed_exchange="DELAYED.EXCHANGE";
    //延迟队列
    public static final String delayed_queue="DELAYED.QUEUE";
    //路由
    public static final String delayed_routingKey="delayed.routingKey";

        //声明交换机
    @Bean
    public CustomExchange dalayedExchange(){
        Map map = new HashMap<>();
        map.put("x-delayed-type","direct");
        
        return new CustomExchange(delayed_exchange,"x-delayed-message",true,false,map);
    }
    //声明队列
    @Bean
    public Queue getQueue(){
        return QueueBuilder.durable(delayed_queue).build();
    }
    //绑定
    @Bean
    public Binding queueBindingExchange(@Qualifier("getQueue") Queue queue,@Qualifier("dalayedExchange")CustomExchange customExchange){
        return BindingBuilder.bind(queue).to(customExchange).with(delayed_routingKey).noargs();
    }
}
三:创建生产者发送消息
package com.zy.rabbitMq.controller;

import com.zy.rabbitMq.config.ConfirmConfig;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@Slf4j
@RequestMapping("/confirm")
@RestController
@AllArgsConstructor
public class ConfirmController {
    private RabbitTemplate rabbitTemplate;
    @GetMapping("/send/{msg}")
    public void senConfirmMsg(@PathVariable(name = "msg")String msg){
        CorrelationData correlationData = new CorrelationData("1");
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.CONFIRM_ROUTING_KEY,msg,correlationData);
        log.info("发送的消息是:{},==id:{}",msg,"1");

        CorrelationData correlationData1 = new CorrelationData("2");
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.CONFIRM_ROUTING_KEY+"23","不可路由"+msg,correlationData1);
        log.info("发送错误的交换机的信息:{},==id为:{}",msg,"2");
    }
}
四:消费者接收消息
package com.zy.rabbitMq.delayed;

        import lombok.extern.slf4j.Slf4j;
        import org.springframework.amqp.core.Message;
        import org.springframework.amqp.rabbit.annotation.RabbitListener;
        import org.springframework.stereotype.Component;

        import java.util.Date;


@Component
@Slf4j
public class DelayedConwumer {

    //开始消费基于插件的延迟消息
    @RabbitListener(queues="DELAYED.QUEUE")
    public void getDelayedMsg(Message message){
        String s = new String(message.getBody());
        log.info("当前时间:{},接收到消息是:{}",new Date().toString(),s);
    }

}

注:根据自己业务逻辑配置交换机及队列,并根据不同业务逻辑设置交换机类型
目前常用的交换机类型为:direct fanout topic topic类型目前最强大

rabbitMq如何保证发送的消息不丢失:
1:队列持久化,2:消息持久化,3:生产者端发布确认
如何保证消息被正常的消费:

1:消费者端开启手动应答

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

版权所有 ©2023-2025 051e.com

ICP备案号:京ICP备12030808号