关于rabbitmq:二RabbitMQ消息分发机制

咱们晓得每个音讯的解决工夫是不同的,换句话说音讯的复杂度是不同的,有些音讯很简单,须要很久的工夫,有些音讯很简略,只需耗时一会就能够实现,而在理论状况下如何分配资源,让效率达到最大化,从而实现按能力分配任务,达到物尽其用。这就须要音讯的散发机制。

一、Fair dispatch(偏心散发)

这里咱们创立DistributionSender.java和DistributionReceiver.java来模仿发送者和接收者。
DistributionSender.java

@Component
public class DistributionSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send(int i) {
        // 发送的音讯
        String message = "This is a task, and the complexity is " + i + "。" + StringUtils.repeat(".", i);
        this.rabbitTemplate.convertAndSend("distribu", message);
    }
}

应用默认交换机,队列为“distribu”

    /**
     * 申明distribu队列
     * 
     * @return
     */
    @Bean
    public Queue DistribuQueue() {
        return new Queue("distribu");
    }

java和DistributionReceiver.java

@Component
public class DistributionReceiver {


    /**
     * 消费者A
     * 
     * @param msg
     */
    @SuppressWarnings("deprecation")
    @RabbitListener(queues = "distribu")
    public void processA(Message message) {
        String msg = new String(message.getBody());
        System.out.println(" DistributionReceiverA  : " + msg);
        SimpleDateFormat time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSSS");
        System.out.println(" ProccessingA... at " + time.format(new Date()));

        try {
            for (char ch : msg.toCharArray()) {
                if (ch == '.') {
                    doWork(1000);
                }
            }
        } catch (InterruptedException e) {
        } finally {
            System.out.println(" A Done! at " + time.format(new Date()));
        }
    }

    private void doWork(long time) throws InterruptedException {
        Thread.sleep(time);
    }

}

增加RabbitTest.java文件模仿申请

@Controller
public class RabbitTest{

    @Autowired
    private DistributionSender distributionSender;

    /**
     * 散发机制音讯发送测试
     */
    @GetMapping("/distribu")
    public void distribu() {
        distributionSender.send(3);
    }
}

运行程序,拜访http://localhost:8080/rabbit/…,能够失去上面的打印的信息:

 DistributionReceiverA  : This is a task, and the complexity is 3。...
 ProccessingA... at 2018-05-23 21:29:18:0628
 A Done! at 2018-05-23 21:29:21:0639

从打印的信息能够看出这里就模仿了实现工作须要3秒钟的工夫工作实现。

上面咱们更改发送的音讯数量,在controller控制器外面进行更改,如下:

    /**
     * 散发机制音讯发送测试
     */
    @GetMapping("/distribu")
    public void distribu() {
        for (int i = 0; i < 5; i++) {
            //发送工作复杂度都为1的音讯
            distributionSender.send(1);
        }
    }

模仿发送5条音讯,并且每条发送的音讯的复杂度都是雷同的,复杂度都为1。

而后再在receiver包中DistributionReceiver.java新增一个消费者B,如下:

/**
     * 消费者B
     * 
     * @param msg
     */
    @SuppressWarnings("deprecation")
    @RabbitListener(queues = "distribu")
    public void processB(Message message) {
        String msg = new String(message.getBody());
        System.out.println(" DistributionReceiverB  : " + msg);
        SimpleDateFormat time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSSS");
        System.out.println(" ProccessingB... at " + time.format(new Date()));

        try {
            for (char ch : msg.toCharArray()) {
                if (ch == '.') {
                    doWork(1000);
                }
            }
        } catch (InterruptedException e) {
        } finally {
            System.out.println(" B Done! at " + time.format(new Date()));
        }
    }

再次运行程序,拜访接口,后果如下:

 DistributionReceiverA  : This is a task, and the complexity is 1。.
 ProccessingA... at 2018-05-22 23:23:43:0014
 DistributionReceiverB  : This is a task, and the complexity is 1。.
 ProccessingB... at 2018-05-22 23:23:43:0014
 A Done! at 2018-05-22 23:23:44:0017
 B Done! at 2018-05-22 23:23:44:0017
 DistributionReceiverA  : This is a task, and the complexity is 1。.
 DistributionReceiverB  : This is a task, and the complexity is 1。.
 ProccessingA... at 2018-05-22 23:23:44:0093
 ProccessingB... at 2018-05-22 23:23:44:0093
 A Done! at 2018-05-22 23:23:45:0095
 B Done! at 2018-05-22 23:23:45:0095
 DistributionReceiverB  : This is a task, and the complexity is 1。.
 ProccessingB... at 2018-05-22 23:23:45:0143
 B Done! at 2018-05-22 23:23:46:0148

在音讯雷同,A、B解决能力一样状况下,咱们能够发现A、B简直是同时解决音讯,音讯发送程序为A->B->A->B->B。能够看出这里并没有实现A与B均匀轮询的状况,在最初的状况B执行了两次。

接着当初咱们把A解决能力更改为每个点要Thread.sleep(4000), B为Thread.sleep(1000),就是B的解决能力是A的四倍。运行一下,咱们看一下打印的后果:

DistributionReceiverB  : This is a task, and the complexity is 1。.
 ProccessingB... at 2018-05-22 23:24:48:0623
 DistributionReceiverA  : This is a task, and the complexity is 1。.
 ProccessingA... at 2018-05-22 23:24:48:0623
 B Done! at 2018-05-22 23:24:49:0624
 DistributionReceiverB  : This is a task, and the complexity is 1。.
 ProccessingB... at 2018-05-22 23:24:49:0663
 B Done! at 2018-05-22 23:24:50:0664
 DistributionReceiverB  : This is a task, and the complexity is 1。.
 ProccessingB... at 2018-05-22 23:24:50:0704
 B Done! at 2018-05-22 23:24:51:0709
 DistributionReceiverB  : This is a task, and the complexity is 1。.
 ProccessingB... at 2018-05-22 23:24:51:0748
 A Done! at 2018-05-22 23:24:52:0629
 B Done! at 2018-05-22 23:24:52:0749

当初咱们能够清晰地看到在这里B解决了4条音讯,而A只解决了1条音讯。这里就是按偏心散发的机制来发送音讯的,即按消费者解决能力来散发音讯。
这就是偏心散发。

二、Round-robin dispatch(轮询散发)

相较于偏心散发而言,轮询散发即不去判断消费者的解决速率,也不思考每个工作的时长,依照轮流排序的形式,把工作一一发给消费者,并且是提前一次性调配,并非一个一个调配。

评论

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

这个站点使用 Akismet 来减少垃圾评论。了解你的评论数据如何被处理