关于springboot:SpringBoot-整合-RabbitMQ

28次阅读

共计 6353 个字符，预计需要花费 16 分钟才能阅读完成。

筹备

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);
 }
    }
}

消费者

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

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

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);
 }
}

路由模式

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

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

主程序

@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);
 }
}

主题模式

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

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

主程序

{
    @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
2020-10-09
 0
关于springboot:Spring-Boot工程中如何优雅地处理异常
关于springboot:在Spring-boot中使用Spring-AOP
关于springboot:Lombok的Data类注解导致被JsonProperty注解字段被序列化了两次
关于springboot:Springboot对连接池的整合