筹备

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

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

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

简略模式

主程序

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

生产者

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

消费者

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

主程序

@SpringBootApplicationpublic 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 }}

生产者

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

消费者

@Componentpublic 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 服务器中定义该交换机.

@SpringBootApplicationpublic 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 替换机会向所有绑定的队列发送数据, 而不是有抉择的发送

@Componentpublic 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注解的嵌套)

@Componentpublic 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. 发送音讯时, 指定路由键

主程序

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

生产者

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

消费者

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

生产者

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

消费者

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