关于springboot:SpringBoot-整合-RabbitMQ

筹备

1.新建springboot我的项目
2.增加Spring for Rabbitmq依赖
3.yml中配置rabbitmq信息:

spring:
  rabbitmq:
    host: 192.168.64.140
    username: admin
    password: admin

每个模式创立一个包,包中蕴含主程序(main办法),生产者,消费者,每测试一个程序启动一个main办法即可.

简略模式

主程序

@SpringBootApplication
public class Main {
    @Autowired
 private Producer p;
 public static void main(String[] args) {
        SpringApplication.run(Main.class,args);
 }
    /*
 springboot我的项目残缺启动实现
 以后对象中须要注入的对象注入实现之后才会执行该办法
 */ @PostConstruct
 public void test(){
        p.send();
 }
    /*
 封装队列参数的对象
 RabbitAutoConfiguraion 主动配置类,会主动发现Queue
 在服务器上定义该队列
 */ @Bean
 public Queue helloworldQueue(){
        //return new Queue("helloworld");//true,false,false 长久,非独占,非主动删除
 return new Queue("helloworld",false);
 }
}

生产者

@Component
public class Producer {
    /*
 AmqpTemplate的实例是在
 RabbitmqAutoConfiguration主动配置类中创立的
 */ @Autowired
 private AmqpTemplate t;
 public void send(){
        t.convertAndSend("helloworld", "Hello World!");
 }
}

消费者

@Component
public class Consumer {
    /*
 消费者主动注册,主动连贯服务器,主动开启音讯监听
 */ @RabbitListener(queues = "helloworld")
    public void receive(String msg){
        System.out.println("收到:"+msg);
 }
}

工作模式

工作模式还须要实现两点:
1.正当散发:
手动ack:springboot整合rabbitmq默认手动ack,并且主动返回回执
qos=1:yml中配置:spring.rabbitmq.listener.simple.prefetch:1
2.长久化:
队列长久化/音讯长久化:springboot整合rabbitmq中都是默认长久的
若是不想要其长久 –>

应用 MessagePostProcessor 前置处理器参数
从音讯中获取音讯的属性对象
在属性中把 DeliveryMode 设置为非长久化

//如果须要设置音讯为非长久化,能够获得音讯的属性对象,批改它的deliveryMode属性
t.convertAndSend("task_queue", (Object) s, new MessagePostProcessor() {
    @Override
    public Message postProcessMessage(Message message) throws AmqpException {
        MessageProperties props = message.getMessageProperties();
        props.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
        return message;
    }
});

主程序

@SpringBootApplication
public class Main {
    @Autowired
 private Producer p;
 @PostConstruct
 public void test(){
        new Thread(new Runnable() {
            @Override
 public void run() {
                p.send();
 }
        }).start();
 //new Thread(() -> p.send()).start();
 }
    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
 }
    @Bean
 public Queue taskQueue(){
        return new Queue("task_queue");//true,false,false
 }
}

生产者

@Component
public class Producer {
    @Autowired
 private AmqpTemplate t;
 public void send(){
        while(true){
            System.out.println("nn输出音讯:");
 String msg = new Scanner(System.in).nextLine();
 t.convertAndSend("task_queue", (Object) msg, new MessagePostProcessor() {
                @Override
 public Message postProcessMessage(Message message) throws AmqpException {
                    MessageProperties p = message.getMessageProperties();
 p.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
 return message;
 }
            });
 }
    }
}

消费者

@Component
public class Consumer {
    @RabbitListener(queues = "task_queue")
    public void receive1(String msg){
        System.out.println("消费者1收到:"+msg);
 }
    @RabbitListener(queues = "task_queue")
    public void receive2(String msg){
        System.out.println("消费者2收到:"+msg);
 }
}

公布订阅模式

主程序

创立 FanoutExcnahge 实例, 封装 fanout 类型交换机定义信息.

spring boot 的主动配置类会主动发现交换机实例, 并在 RabbitMQ 服务器中定义该交换机.

@SpringBootApplication
public class Main {
    @Autowired
 private Producer p;
 @PostConstruct
 public void test(){
        new Thread(new Runnable() {
            @Override
 public void run() {
                p.send();
 }
        }).start();
 }
    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
 }
    //定义fanout交换机:logs
 @Bean
 public FanoutExchange logs(){
        //return new FanoutExchange("logs");//默认true,false
 return new FanoutExchange("logs",false,false);
 }
}

生产者

生产者向指定的交换机 logs 发送数据.

不须要指定队列名或路由键, 即便指定也有效, 因为 fanout 替换机会向所有绑定的队列发送数据, 而不是有抉择的发送

@Component
public class Producer {
    @Autowired
 private AmqpTemplate t;
 public void send(){
        while (true){
            System.out.println("输出音讯:");
 String msg=new Scanner(System.in).nextLine();
 t.convertAndSend("logs", "",msg);
 }
    }
}

消费者

消费者须要执行以下操作:

定义随机队列(随机命名,非长久,排他,主动删除)
定义交换机(能够省略, 已在主程序中定义)
将队列绑定到交换机

spring boot 通过注解实现以上操作:(留神@RabbitListener注解的嵌套)

@Component
public class Consumer {
    /*
 1.随即队列
 2.指定绑定的交换机
 */ @RabbitListener(bindings = @QueueBinding(
            value = @Queue,//服务器主动取名,false,true,true
 exchange = @Exchange(name = "logs",declare = "false")//declare指定的是一个已定义过的交换机,而不是从新定义
 ))
    public void reveice1(String msg){
        System.out.println("消费者1收到:"+msg);
 }
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,//服务器主动取名,false,true,true
 exchange = @Exchange(name = "logs",declare = "false")
    ))
    public void reveice2(String msg){
        System.out.println("消费者2收到:"+msg);
 }
}

路由模式

与公布和订阅模式代码相似, 只是做以下三点调整:

应用 direct 交换机
队列和交换机绑定时, 设置绑定键
发送音讯时, 指定路由键

主程序

@SpringBootApplication
public class Main {
    @Autowired
 private Producer p;
 @PostConstruct
 public void test(){
        new Thread(new Runnable() {
            @Override
 public void run() {
                p.send();
 }
        }).start();
 }
    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
 }
    //定义direct交换机:logs
 @Bean
 public DirectExchange logs(){
        //return new FanoutExchange("logs");//默认true,false
 return new DirectExchange("direct_logs",false,false);
 }
}

生产者

@Component
public class Producer {
    @Autowired
 private AmqpTemplate t;
 public void send(){
        while (true){
            System.out.println("输出音讯:");
 String msg=new Scanner(System.in).nextLine();
 System.out.println("输出路由键:");
 String key=new Scanner(System.in).nextLine();
 t.convertAndSend("direct_logs", key,msg);
 }
    }
}

消费者

@Component
public class Consumer {
    /*
 1.随即队列
 2.指定绑定的交换机
 */ @RabbitListener(bindings = @QueueBinding(
            value = @Queue,//服务器主动取名,false,true,true
 exchange = @Exchange(name = "direct_logs",declare = "false"),//declare指定的是一个已定义过的交换机,而不是从新定义
 key = "error"
 ))
    public void reveice1(String msg){
        System.out.println("消费者1收到:"+msg);
 }
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,//服务器主动取名,false,true,true
 exchange = @Exchange(name = "direct_logs",declare = "false"),
 key = {"info","warning","error"}
    ))
    public void reveice2(String msg){
        System.out.println("消费者2收到:"+msg);
 }
}

主题模式

主题模式不过是具备非凡规定的路由模式, 代码与路由模式基本相同, 只做如下调整:

应用 topic 交换机
应用非凡的绑定键和路由键规定

主程序

{
    @Autowired
 private Producer p;
 @PostConstruct
 public void test(){
        new Thread(new Runnable() {
            @Override
 public void run() {
                p.send();
 }
        }).start();
 }
    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
 }
    //定义topic交换机:logs
 @Bean
 public TopicExchange logs(){
        //return new TopicExchange("logs");//默认true,false
 return new TopicExchange("topic_logs",false,false);
 }
}

生产者

@Component
public class Producer {
    @Autowired
 private AmqpTemplate t;
 public void send(){
        while (true){
            System.out.println("输出音讯:");
 String msg=new Scanner(System.in).nextLine();
 System.out.println("输出路由键:");
 String key=new Scanner(System.in).nextLine();
 t.convertAndSend("topic_logs", key,msg);
 }
    }
}

消费者

@Component
public class Consumer {
    /*
 1.随即队列
 2.指定绑定的交换机
 */ @RabbitListener(bindings = @QueueBinding(
            value = @Queue,//服务器主动取名,false,true,true
 exchange = @Exchange(name = "topic_logs",declare = "false"),//declare指定的是一个已定义过的交换机,而不是从新定义
 key = "*.orange.*"
 ))
    public void reveice1(String msg){
        System.out.println("消费者1收到:"+msg);
 }
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,//服务器主动取名,false,true,true
 exchange = @Exchange(name = "topic_logs",declare = "false"),
 key = {"*.*.rabbit","lazy.#"}
    ))
    public void reveice2(String msg){
        System.out.println("消费者2收到:"+msg);
 }
}

关于springboot:SpringBoot-整合-RabbitMQ

筹备

简略模式

主程序

生产者

消费者

工作模式

主程序

生产者

消费者

公布订阅模式

主程序

生产者

消费者

路由模式

主程序

生产者

消费者

主题模式

主程序

生产者

消费者

评论

发表回复取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

关于springboot:SpringBoot-整合-RabbitMQ

筹备

简略模式

主程序

生产者

消费者

工作模式

主程序

生产者

消费者

公布订阅模式

主程序

生产者

消费者

路由模式

主程序

生产者

消费者

主题模式

主程序

生产者

消费者

评论

发表回复 取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

发表回复取消回复