RabbitMQ producer consumer code case, super detailed!

[Chen Xi to work hard] : hello, I am Chen Xi, very glad you can read, nickname is to hope that they can continue to improve, toward excellent programmers! The blog comes from the summary of the problems encountered in the project and programming, and occasionally THERE will be reading and sharing. I will update the Summary of relevant knowledge points such as Java front end, background, database and project case successively. Thank you for your reading and attention. We quarrymen, with the heart of the cathedral, may we go to our respective loves…

First, the original intention of the article

Beginner RabbitMQ network cases are scattered, in order to facilitate beginners better learning, sorting out this article, the code is completely real, beginners to practice!

Graph TD RabbitMQ --> Case study

To learn about message queues for the first time, it is important to understand how they are used. Why? And how to use it? In order to share this article to beginners to provide him with a detailed case, the following code can be directly practiced!

The three core points of message queuing: decoupling, asynchro, and peak clipping

Second, code details

The relevant code for the producer

/** * The producer that wrote the message */
@Component
public class MsgProducer implements RabbitTemplate.ConfirmCallback {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    / / due to the scope attribute is set to ConfigurableBeanFactory rabbitTemplate SCOPE_PROTOTYPE, so can't automatic injection
    private RabbitTemplate rabbitTemplate;
    /** * the constructor inserts rabbitTemplate */
    @Autowired
    public MsgProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        rabbitTemplate.setConfirmCallback(this); // If rabbitTemplate is a singleton, the callback is set last
    }
     /** * A switch with a route with a queue *@param content
     */
    public void sendMsg(String content) {
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        // Put the message in the queue corresponding to ROUTINGKEY_A, which is queue A
        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_A, RabbitConfig.ROUTINGKEY_A, content, correlationId);
        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_B, RabbitConfig.ROUTINGKEY_B, content, correlationId);
    }

    /** * Broadcast mode *@param content
     */
    public void sendAll(String content) {
        rabbitTemplate.convertAndSend("fanoutExchange"."", content);
    }

    /** * callback */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        //logger.info(" callback id:" + correlationData);
        if (ack) {
            logger.info("Message success consumption");
        } else {
            logger.info("Message consumption failed :"+ cause); }}}Copy the code

Code for the Controller layer

@RestController
public class SendController {

    @Autowired
    private MsgProducer msgProducer;

    @RequestMapping(value = "/send",method = RequestMethod.GET)
    public void send(int length){
        for (int i=1; i<=length; i++){ msgProducer.sendMsg("This is the number I sent."+i+"A message"); }}@RequestMapping(value = "/sendAll",method = RequestMethod.GET)
    public void sendAll(int length){
        for (int i=1; i<=length; i++){ msgProducer.sendAll("This is the number I sent."+i+"A message"); }}}Copy the code

The relevant code of the consumer

@Component
@RabbitListener(queues = RabbitConfig.QUEUE_A)
public class MsgReceiver_one {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @RabbitHandler
    public void process(String content) {
        logger.info("Consumer one receives the message in processing queue A:"+ content); }}Copy the code

These two listen on queues A and B, respectively

@Component
@RabbitListener(queues = RabbitConfig.QUEUE_B)
public class MsgReceiver_two {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @RabbitHandler
    public void process(String content) {
        logger.info("Consumer two receives the message in processing queue A:"+ content); }}Copy the code

Test: Enter the relevant parameters in the browser to access the first path

The first path here is the producer to queue A and queue B

See the console print the relevant information

Test: enter access to the second path in the browser and enter related parameters

Here the second path is published in broadcast form and can be consumedSee the console print the relevant information

http://localhost:15672/#/queues

Information about the project configuration file

@Configuration
public class RabbitConfig {

    @Value("${spring.rabbitmq.host}")
    private String host;

    @Value("${spring.rabbitmq.port}")
    private int port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    public static final String FANOUT_EXCHANGE="fanoutExchange";

    / / switches
    public static final String EXCHANGE_A = "my-mq-exchange_A";
    public static final String EXCHANGE_B = "my-mq-exchange_B";
    public static final String EXCHANGE_C = "my-mq-exchange_C";
    / / the queue
    public static final String QUEUE_A = "QUEUE_A";
    public static final String QUEUE_B = "QUEUE_B";
    public static final String QUEUE_C = "QUEUE_C";
    // Route key
    public static final String ROUTINGKEY_A = "spring-boot-routingKey_A";
    public static final String ROUTINGKEY_B = "spring-boot-routingKey_B";
    public static final String ROUTINGKEY_C = "spring-boot-routingKey_C";

    @Bean
    public ConnectionFactory connectionFactory(a) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host,port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        //connectionFactory.setVirtualHost("/test");
        connectionFactory.setPublisherConfirms(true);
        return connectionFactory;
    }

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    // Must be of type prototype
    public RabbitTemplate rabbitTemplate(a) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        return template;
    }

    /** * Consumer configuration * 1. Set the switch type * 2. Bind queues to switch FanoutExchange: Distribute messages to all bound queues without the concept of a routingkey Matching DirectExchange by adding the key-value attribute: routingKey to the specified queue TopicExchange: multiple keyword matching */
    @Bean
    public DirectExchange defaultExchangeA(a) {
        return new DirectExchange(EXCHANGE_A);
    }

    @Bean
    public DirectExchange defaultExchangeB(a) {
        return new DirectExchange(EXCHANGE_B);
    }

    @Bean
    public DirectExchange directExchangeC(a){
        return new DirectExchange(EXCHANGE_C);
    }

    /** * get queue A *@return* /
    @Bean
    public Queue queueA(a) {
        return new Queue(QUEUE_A, true); // Queue persistence
    }

    @Bean
    public Queue queueB(a) {
        return new Queue(QUEUE_B, true); // Queue persistence
    }

    @Bean
    public Queue queueC(a) {
        return new Queue(QUEUE_C, true); // Queue persistence
    }

    /** * queue bind switch *@return* /
    @Bean
    public Binding bindingA(a) {
        return BindingBuilder.bind(queueA()).to(defaultExchangeA()).with(RabbitConfig.ROUTINGKEY_A);
    }

    @Bean
    public Binding bindingB(a){
        return BindingBuilder.bind(queueB()).to(defaultExchangeB()).with(RabbitConfig.ROUTINGKEY_B);
    }

    @Bean
    public Binding bindingC(a){
        return BindingBuilder.bind(queueC()).to(directExchangeC()).with(RabbitConfig.ROUTINGKEY_C);
    }

    / / configuration fanout_exchange
    // Fanout supports only unified broadcasts
    @Bean
    FanoutExchange fanoutExchange(a) {
        return new FanoutExchange(RabbitConfig.FANOUT_EXCHANGE);
    }

    // Bind all queues to the switch
    @Bean
    Binding bindingExchangeA(FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queueA()).to(fanoutExchange);
    }
    @Bean
    Binding bindingExchangeB(FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queueB()).to(fanoutExchange);
    }
    @Bean
    Binding bindingExchangeC(FanoutExchange fanoutExchange) {
        returnBindingBuilder.bind(queueC()).to(fanoutExchange); }}Copy the code

Summary of RabbitMQ case

1.First create a configuration class and then create the queue2.Create the producer object that the producer wants to use the message queue, and then pass the message you want to pass3.Create the control layer fills in the path, creates the producer object, and sends the message4.Consumer, fill in the name of the consumption queue, listen ready to start consumptionCopy the code

Thank you so much for reading this, if this article has been helpful to you, please leave a like 👍 follow ❤️ share 👥 comment 💬 Thanks!!