当前位置：首页 > news >正文

SpringBoot RabbitMQ 集成二 Work queues

news 来源：原创 2024/4/29 4:20:03

为什么80%的码农都做不了架构师？>>>

上一节的一个工程中两个类，一个负责发送，一个负责接收，也就是一一对于的关系。

只要消息发出了，接收者就处理；当接收效率较低时，就会出现接收者处理不过来，我们就可能会处理不过来，于是我们就可能多配置接受者。这个模式就是"Work queues",它的结构如下

20170404002111089.

多个接收者，它们会出现什么情况呢？是否像大锅饭，有的人撑死，有的人饿死。这个通过例子验证。

一、编写代码

1、编写常量类RabbitWorkQueuesConstant

package com.lvgang.springbootrabbitmq.workqueues;

/**
 * @author lvgang
 */
public class RabbitWorkQueuesConstant {

    public  static final  String QUEUQ = "Queue_WorkQueues";

}

2、编写配置类RabbitWorkQueuesConfig

package com.lvgang.springbootrabbitmq.workqueues;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author lvgang
 */
@Configuration
public class RabbitWorkQueuesConfig {

    /**
        * Queue 可以有4个参数
     *      1.队列名
     *      2.durable       持久化消息队列 ,rabbitmq重启的时候不需要创建新的队列 默认true
     *      3.auto-delete   表示消息队列没有在使用时将被自动删除 默认是false
     *      4.exclusive     表示该消息队列是否只在当前connection生效,默认是false
     */

    @Bean
    public Queue createWorkQueuesQueue() {
        return new Queue(RabbitWorkQueuesConstant.QUEUQ,true);
    }

}

3、编写消息生产者WorkQueuesSender

package com.lvgang.springbootrabbitmq.workqueues;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.UUID;

/**
 * @author lvgang
 */
@Component
public class WorkQueuesSender {
	private static Logger logger = LoggerFactory.getLogger(WorkQueuesSender.class);

	@Autowired
	private RabbitTemplate rabbitTemplate;

	public void send(int i) {
		CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
		String message = "WorkQueues="+ i+"," + new Date() + ", content= " + UUID.randomUUID().toString();
		rabbitTemplate.convertAndSend(RabbitWorkQueuesConstant.QUEUQ,(Object) message,correlationData);
		logger.info("send ok,"+new Date()+","+message);
	}
}

4、编写消息接收者1WorkQueuesReceiver1

package com.lvgang.springbootrabbitmq.workqueues;

import com.lvgang.springbootrabbitmq.helloworld.HelloWorldSender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @author lvgang
 */
@Component
@RabbitListener(queues = RabbitWorkQueuesConstant.QUEUQ)
public class WorkQueuesReceiver1 {
    private static Logger logger = LoggerFactory.getLogger(HelloWorldSender.class);
    @RabbitHandler
    public void process(String message) {
        logger.info("Receiver1 : " + message +","+ new Date());
    }
}

5、编写消息接收者2WorkQueuesReceiver2

package com.lvgang.springbootrabbitmq.workqueues;

import com.lvgang.springbootrabbitmq.helloworld.HelloWorldSender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @author lvgang
 */
@Component
@RabbitListener(queues = RabbitWorkQueuesConstant.QUEUQ)
public class WorkQueuesReceiver2 {
    private static Logger logger = LoggerFactory.getLogger(HelloWorldSender.class);
    @RabbitHandler
    public void process(String message) {
        logger.info("Receiver2 : " + message +","+ new Date());
    }
}

二、测试结果

1、编写测试类WorkQueuesTests

package com.lvgang.springbootrabbitmq.workqueues;


import com.lvgang.springbootrabbitmq.helloworld.HelloWorldSender;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest
public class WorkQueuesTests {
	private static Logger logger = LoggerFactory.getLogger(HelloWorldSender.class);
	@Autowired
	private WorkQueuesSender workQueuesSender;

	@Test
	public void hello() throws Exception {
		int i=1;
		while (true) {
			try {
				if(i<=10) {
					workQueuesSender.send(i);
				}
				i++;
				Thread.sleep(1000);
			} catch (Exception e) {
				logger.error(e.getMessage(),e);
			}
		}
	}
}