spring boot 中使用 RabbitMQ 教程二 wor

2018-02-09 本文已影响2392人阿波罗程序猿

在spring boot 中使用 RabbitMQ 教程一生产者、队列、消费者，中讲的是，一对一的关系。这次我们来探寻一对多的关系。

RabbitMQ给我提供了类似一对多的关系，就是多个相同的消费者，来消费同一个队列中的消息。使用多个消费者来消费同一个队列中的消息，使这些消息将会平均分到各个消费者中进行消费。

一对多
基本的例如启动rabbitmq-server和配置application.properties这里就不谈了。请查看上期内容。
定义一个队列workQueue，两个消费者workReceiver 、workReceiver 1，一个生产者workSend。

import com.zb.rabbitMQtest.workqueues.receiver.WorkReceiver;
import com.zb.rabbitMQtest.workqueues.send.WorkSend;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author 张博
 */
@Configuration
public class RabbitMQConfig {

    @Bean
    public Queue workQueue() {
        return new Queue("work-queue");
    }

    @Bean
    public WorkReceiver workReceiver() {
        return new WorkReceiver("Receiver0");
    }

    @Bean
    public WorkReceiver workReceiver1() {
        return new WorkReceiver("Receiver1");
    }

    @Bean
    public WorkSend workSend() {
        return new WorkSend();
    }
}

定义消费者类。使用@RabbitListener，监听名字为work-queue的队列，接收到消息后，会把消息循环分发到，之前定义的两个消费者对象上。

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author 张博
 */
@Component
public class WorkReceiver {

    private String receiverInstance;

    public WorkReceiver(String receiverInstance) {
        this.receiverInstance = receiverInstance;
    }

    @RabbitListener(queues = "work-queue")
    public void receive(String str) {
        System.out.println(receiverInstance.concat(" =====: ").concat(str));
    }
}

定义生产者类，使用RabbitTemplate提供的convertAndSend方法一次性发送10条消息到队列中。

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @author 张博
 */
@Component
public class WorkSend {

    @Autowired
    private Queue workQueue;
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void send() {
        for (int i = 0; i < 10; i++) {
            String msg = "你好 " + i;
            System.out.println("send =====:".concat(msg));
            rabbitTemplate.convertAndSend(workQueue.getName(), msg);
        }
    }
}

测试

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

/**
 * @author 张博
 */
@RunWith(SpringRunner.class)
@SpringBootTest
public class WorkSendTest {

    @Autowired
    private WorkSend workSend;

    @Test
    public void send() throws Exception {
        workSend.send();
    }
}

运行结果，spring-amqp 默认采用的是公平调度（Fair dispatch）的形式。

send =====:你好 0
send =====:你好 1
send =====:你好 2
send =====:你好 3
send =====:你好 4
send =====:你好 5
send =====:你好 6
send =====:你好 7
send =====:你好 8
send =====:你好 9

Receiver1 =====: 你好 1
Receiver0 =====: 你好 0
Receiver0 =====: 你好 2
Receiver1 =====: 你好 3
Receiver0 =====: 你好 4
Receiver1 =====: 你好 5
Receiver0 =====: 你好 6
Receiver1 =====: 你好 7
Receiver0 =====: 你好 8
Receiver1 =====: 你好 9

公平调度或循环调度。
默认时RabbitMQ将每条发送的消息都将由下一个消费者进行消费，就是说每个消费者都能与其它消费者接收到相同的数量的消息。但是问题来了，假如A消费者接到消息处理时间很长，B消费者接收到的消息处理时间很短，那么B消费者大部分时间将处于空闲状态下。这样时间一长性能会下降，至少看起来没那么均匀了。这种做法就是“循环调度”。RabbitMQ只是当消息进入到队列中时进行消息的分发调度。并没有考虑到消费者具体完成消费的数量。
而且在spring-amqp默认采用的是公平调度。并不是循环调度，那种只注意分发同等消息数量给消费者的情况。如果想使用循环调度的话，那么设置SimpleMessageListenerContainer中的DEFAULT_PREFETCH_COUNT = 0即可。DEFAULT_PREFETCH_COUNT = 1是公平调度。请看这里。
根据上面的例子把公平调度修改成循环调度factory.setPrefetchCount(0);

import com.zb.rabbitMQtest.workqueues.receiver.WorkReceiver;
import com.zb.rabbitMQtest.workqueues.send.WorkSend;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author 张博
 */
@Configuration
public class RabbitMQConfig {

    @Bean
    public Queue workQueue() {
        return new Queue("work-queue");
    }

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory());
        factory.setPrefetchCount(0);
        return factory;
    }

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory cachingConnectionFactory =  new CachingConnectionFactory("127.0.0.1", 5672);
        cachingConnectionFactory.setUsername("guest");
        cachingConnectionFactory.setPassword("guest");
        return cachingConnectionFactory;
    }

    @Bean
    public WorkReceiver workReceiver() {
        return new WorkReceiver("Receiver0");
    }

    @Bean
    public WorkReceiver workReceiver1() {
        return new WorkReceiver("Receiver1");
    }

    @Bean
    public WorkSend workSend() {
        return new WorkSend();
    }
}

为@RabbitListener添加属性containerFactory创建指定监听此队列上的一个消息监听容器。

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author 张博
 */
@Component
public class WorkReceiver {

    private String receiverInstance;

    public WorkReceiver(String receiverInstance) {
        this.receiverInstance = receiverInstance;
    }

    @RabbitListener(queues = "work-queue", containerFactory = "rabbitListenerContainerFactory")
    public void receive(String str) {
        System.out.println(receiverInstance.concat(" =====: ").concat(str));
    }
}

prefetchCount设置为1时表示，你不要给我发新的消息了，等我处理完你再给我发新的消息。这时它会把这个消息交给空闲的消费者进行处理。这样看起来很公平。
公平调度

spring boot 中使用 RabbitMQ 教程二 wor

导航

猜你喜欢

热点阅读