4. 基于配置的消费者实现

一灰灰blog开源RabbitMQRabbitMQSpring约 1621 字大约 5 分钟

RabbitMQ基础教程之基于配置的消费者实现

相关博文,推荐查看:

  1. RabbitMq基础教程之安装与测试open in new window
  2. RabbitMq基础教程之基本概念open in new window
  3. RabbitMQ基础教程之基本使用篇open in new window
  4. RabbitMQ基础教程之使用进阶篇open in new window
  5. RabbitMQ基础教程之Spring&JavaConfig使用篇open in new window
  6. RabbitMQ基础教程之Spring-JavaConfig-FactoryBean使用姿势

前面一篇介绍了使用工厂方式创建消费者,其中一个不太友好的地方就在配置都是硬编码的方式,不太灵活,那么是否可以结合前一篇的FactoryBean来实现从配置中来灵活的创建消费者呢?

I. 动态配置实现消费者程序

1. 配置文件加载

首先就是需要从配置文件中获取相应的配置信息,借助JavaConfig,加一个注解即可

@Configuration
@PropertySource("classpath:dynamicConfig.properties")
public class DynSpringConfig {

    @Autowired
    private Environment environment;

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setHost(environment.getProperty("dyn.mq.host"));
        factory.setPort(Integer.parseInt(environment.getProperty("dyn.mq.port")));
        factory.setUsername(environment.getProperty("dyn.mq.uname"));
        factory.setPassword(environment.getProperty("dyn.mq.pwd"));
        factory.setVirtualHost(environment.getProperty("dyn.mq.vhost"));
        return factory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }
}

主要就是 @PropertySource("classpath:dynamicConfig.properties") , 表示从dynamicConfig.properties文件中读取相应的配置,而这些配置,会存放在 Environment 容器内;

获取配置的方式,就是通过org.springframework.core.env.PropertyResolver#getProperty(java.lang.String)获取

2. 消费者通用实现

实现一个简单的通用的消费端,主要根据前一篇博文中定义的MQContainerFactory,来生成SimpleMessageListenerContainer,然后注入消费服务,并启动容器

public class DynamicConsumer {
    public DynamicConsumer(MQContainerFactory fac) throws Exception {
        SimpleMessageListenerContainer container = fac.getObject();
        container.setMessageListener(new AbsMQConsumer() {
            @Override
            public boolean process(Message message, Channel channel) {
                System.out.println("DynamicConsumer: " + fac.getQueue() + " | " + new String(message.getBody()));
                return true;
            }
        });

        container.start();
    }
}

上面是一个非常简单的实现,针对常见的的RabbitMQ消息消费而言,也可以写一个泛型类,然后借助Spring的事件机制,实现一个通用的消费端,一种case如下:

public class JsonMsgConsumer {
    public JsonMsgConsumer(ApplicationContext apc, MQContainerFactory fac, Class<?> msgType) throws Exception {
        SimpleMessageListenerContainer container = fac.getObject();
        container.setMessageListener(new AbsMQConsumer() {
            @Override
            public boolean process(Message message, Channel channel) {
                System.out.println("DynamicConsumer: " + fac.getQueue() + " | " + new String(message.getBody()));

                Object type = JSONObject.parseObject(message.getBody(), msgType);
                apc.publishEvent(type);
                return true;
            }
        });

        container.start();
    }
}

如果message中的数据,是通过Json序列化方式存入,则使用方,只需要监听对应的Event消费数据即可,完全不用再关系消费端的情况了

3. MQContainerFactory 初始化

根据配置文件中的信息,初始化factory,这个可谓是最关键的地方了,实现也和之前大致类似,只不过是将硬编码改成配置信息读取而已,完整的配置文件如下

@Configuration
@PropertySource("classpath:dynamicConfig.properties")
public class DynSpringConfig {

    @Autowired
    private Environment environment;

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setHost(environment.getProperty("dyn.mq.host"));
        factory.setPort(Integer.parseInt(environment.getProperty("dyn.mq.port")));
        factory.setUsername(environment.getProperty("dyn.mq.uname"));
        factory.setPassword(environment.getProperty("dyn.mq.pwd"));
        factory.setVirtualHost(environment.getProperty("dyn.mq.vhost"));
        return factory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public AmqpProducer amqpProducer() {
        return new AmqpProducer();
    }


    @Bean
    public DynamicConsumer dynamicConsumer(ConnectionFactory connectionFactory, RabbitAdmin rabbitAdmin)
            throws Exception {
        MQContainerFactory fac = MQContainerFactory.builder().directExchange(environment.getProperty("dyn.mq.exchange"))
                .queue(environment.getProperty("dyn.mq.queue"))
                .autoDeleted(Boolean.parseBoolean(environment.getProperty("dyn.mq.autoDeleted")))
                .autoAck(Boolean.parseBoolean(environment.getProperty("dyn.mq.autoAck")))
                .durable(Boolean.parseBoolean(environment.getProperty("dyn.mq.durable")))
                .routingKey(environment.getProperty("dyn.mq.routingKey")).rabbitAdmin(rabbitAdmin)
                .connectionFactory(connectionFactory).build();

        return new DynamicConsumer(fac);
    }
}

4. 测试

配置文件内容:

dyn.mq.host=127.0.0.1
dyn.mq.port=5672
dyn.mq.uname=admin
dyn.mq.pwd=admin
dyn.mq.vhost=/
dyn.mq.exchange=fac.direct.exchange
dyn.mq.queue=dyn.queue
dyn.mq.durable=true
dyn.mq.autoDeleted=false
dyn.mq.autoAck=false
dyn.mq.routingKey=fac-routing

测试方法

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(classes = DynSpringConfig.class)
public class DynamicConsumerUnit {
    @Autowired
    private AmqpProducer amqpProducer;

    @Test
    public void testDirectConsumer() throws InterruptedException {
        String[] routingKey = new String[]{"hello.world", "fac-routing", "test1"};
        for (int i = 0; i < 100; i++) {
            amqpProducer.publishMsg("fac.direct.exchange", routingKey[i % 3],
                    ">>> hello " + routingKey[i % 3] + ">>> " + i);
        }
        System.out.println("-------over---------");

        Thread.sleep(1000 * 60 * 10);
    }
}

执行之后,就可以看到正常的消费了

5. 扩充与小结

看完之后,可能有一个问题,为什么要这样做,好处是什么?

大部分的时候,从MQ获取消息的逻辑都一样,唯一的区别在于获取到数据之后做的业务而言,如果把这一块完全的抽象出来,通过配置的方式,那么额外的新增mq的消费,就不需要再改消费端的代码了,然后就会有一个疑问,上面的配置文件中,生成dynamicConsumer的bean不也是需要额外写么?

如果将配置信息,以某种数组的方式定义,遍历读取这些配置,然后创建多个DynamicConsuer实例,是否就能支持动态扩展呢?

将配置改成下面的进行尝试

@Configuration
@PropertySource("classpath:dynamicConfig.properties")
public class DynSpringConfig {

    @Autowired
    private Environment environment;

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setHost(environment.getProperty("dyn.mq.host"));
        factory.setPort(Integer.parseInt(environment.getProperty("dyn.mq.port")));
        factory.setUsername(environment.getProperty("dyn.mq.uname"));
        factory.setPassword(environment.getProperty("dyn.mq.pwd"));
        factory.setVirtualHost(environment.getProperty("dyn.mq.vhost"));
        return factory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public AmqpProducer amqpProducer() {
        return new AmqpProducer();
    }


    @Autowired
    private ConnectionFactory connectionFactory;

    @PostConstruct
    public void dynamicConsumer()
            throws Exception {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        MQContainerFactory fac = MQContainerFactory.builder().directExchange(environment.getProperty("dyn.mq.exchange"))
                .queue(environment.getProperty("dyn.mq.queue"))
                .autoDeleted(Boolean.parseBoolean(environment.getProperty("dyn.mq.autoDeleted")))
                .autoAck(Boolean.parseBoolean(environment.getProperty("dyn.mq.autoAck")))
                .durable(Boolean.parseBoolean(environment.getProperty("dyn.mq.durable")))
                .routingKey(environment.getProperty("dyn.mq.routingKey")).rabbitAdmin(rabbitAdmin)
                .connectionFactory(connectionFactory).build();

        new DynamicConsumer(fac);
    }
}

注意之前 dynamicConsumer 是bean的创建,改成了初始化一个实例,如果配置文件是数组,内部用一个遍历就可以全部加载,现在就需要验证上面的配置改动之后,是否依然可以消费数据

实测ok,部分输出如下

-------over---------
DynamicConsumer: dyn.queue | >>> hello fac-routing>>> 1
DynamicConsumer: dyn.queue | >>> hello fac-routing>>> 4
DynamicConsumer: dyn.queue | >>> hello fac-routing>>> 7

II. 其他

项目地址

一灰灰Blogopen in new windowhttps://liuyueyi.github.io/hexblogopen in new window

一灰灰的个人博客,记录所有学习和工作中的博文,欢迎大家前去逛逛

声明

尽信书则不如,已上内容,纯属一家之言,因个人能力有限,难免有疏漏和错误之处,如发现bug或者有更好的建议,欢迎批评指正,不吝感激

扫描关注

QrCode
QrCode
Loading...