码迷,mamicode.com
首页 > 编程语言 > 详细

SpringBoot Rabbitmq发送消息

时间:2019-02-27 17:34:42      阅读:366      评论:0      收藏:0      [点我收藏+]

标签:artifact   fanout   oid   tps   hang   color   doc   autowired   config   

官方文档:https://docs.spring.io/spring-boot/docs/2.1.3.RELEASE/reference/htmlsingle/#boot-features-amqp

引入依赖:

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

发送消息代码:

@RestController
@RequestMapping("/")
public class SenderMsgController {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @RequestMapping(value = "/{str}")
    public void testSend(@RequestParam("str") String str) throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            int millis = 500;
            Thread.sleep(new Long(millis));
            if (i%2 == 1) {
                String isr = "{\n" + "\t\"dd\":" + i + "}";

                amqpTemplate.convertAndSend("DirectExchange","test.1",isr,new MyMessageConverter());

            }else{
                String isr = "{\n" + "\t\"dd\":" + i + "}";
                amqpTemplate.convertAndSend("DirectExchange","test.2",isr,new MyMessageConverter());
            }
            System.out.println("第"+i+"次发送");
        }
    }
}
MyMessageConverter:
//MessagePostProcessor 接口可以对发送请求之前的Message 进行操作,这里我设置了contenttype为json格式
public class MyMessageConverter implements MessagePostProcessor {
    @Override
    public Message postProcessMessage(Message message) throws AmqpException {
        message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_JSON);
        return message;
    }
}

创建交换机、队列并互相绑定设置路由key

@Component
public class AmqpAdminConfig {

    @Autowired
    public AmqpAdmin amqpAdmin;


    @Bean
    public DirectExchange createDirectExchange(){
        DirectExchange directExchange = new DirectExchange("DirectExchange", false, false);
        amqpAdmin.declareExchange(directExchange);
        return directExchange;
    }

//    @Bean
//    public void createFanoutExchange(){
//        amqpAdmin.declareExchange(new FanoutExchange("FanoutExchange",false,false));
//    }

    @Bean
    public Queue createQueue1(){
        Queue queue = new Queue("queue-1", false, false, false);
        amqpAdmin.declareQueue(queue);
        return queue;
    }

    @Bean
    public Queue createQueue2(){
        Queue queue = new Queue("queue-2", false, false, false);
        amqpAdmin.declareQueue(queue);
        return queue;
    }

    @Bean
    public void createBinding1(){
        Binding bind = BindingBuilder.bind(createQueue1()).to(createDirectExchange()).with("test.1");
        amqpAdmin.declareBinding(bind);
    }
    @Bean
    public void createBinding2(){
        Binding bind = BindingBuilder.bind(createQueue2()).to(createDirectExchange()).with("test.2");
        amqpAdmin.declareBinding(bind);
    }

}

根据官方文档知道AmqpTemplate 和AmqpAdmin  已经自动配置,可直接注入使用,AmqpTemplate 封装了发送与接收的各种操作,AmqpAdmin  封装了针对交换机和消息队列的各种操作

SpringBoot Rabbitmq发送消息

标签:artifact   fanout   oid   tps   hang   color   doc   autowired   config   

原文地址:https://www.cnblogs.com/shiguotao-com/p/10445130.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!