通用的消息队列(redis,kafka,rabbitmq)--生产者篇

星痕

发布于 2020-06-19 16:57:38

5730

发布于 2020-06-19 16:57:38

网上有很多消息队列的中间件，如redis,kafka,rabbitmq,这些都很强大但用起来，每个的用法都不一样，有没有一种办法，我只需要实现一种方法，就能随意使用哪个中间件都可以呢. 我这边设计如下：

生产者通用消息对象，里面只有主题及消息

@Data
@NoArgsConstructor
public class MessageQueueDto {
    public MessageQueueDto(String topic, String message) {
        this.topic = topic;
        this.message = message;
    }


    /**
     * 消息主题
     */
    private String topic;


    /**
     * 消息内容
     */
    private String message;

}

2.生产者接口设计，用于各种消息队列的实现

/**
 * 消息队列生产者
 * @author starmark
 * @date 2020/5/1  上午10:36
 */
public interface IMessageQueueProducerService {

    /**
     * 发布消息队列
     * @param messageQueueDto 消息内容
     * @return 是否发布成功
     */
    boolean produceMessage(MessageQueueDto messageQueueDto);

    /**
     *
     * @param producerType 生产者类型
     * @return 是否支持该生产者
     */
    boolean support(String  producerType);
}

3.生产者工厂的实现，

@Service
public class MessageQueueProducerFactory {
    @Autowired
    private List<IMessageQueueProducerService> messageQueueProducerServices;

    public IMessageQueueProducerService getMessageQueueProducerService(String producerType) {

        IMessageQueueProducerService producerService = messageQueueProducerServices.stream().filter(messageQueueProducerService -> messageQueueProducerService.support(producerType)).findFirst().orElse(null);
        Assert.notNull(producerService, "消息队列生产者不存在");
        return producerService;
    }
}

如果你要发送一个消息，就直接这样调用就可以了示例代码如下：

@Service
@Slf4j
public class LogServiceImpl implements ILogService {


    @Value("${starmark.log.access.kafka.topic:accessLog}")
    private String accessLogTopic;
    @Autowired
    private MessageQueueProducerFactory messageQueueProducerFactory;

    @Override
    @Async
    public void saveLog(AccessLogDto logDto) {

        log.info("请求日志:{}",
                logDto);
        messageQueueProducerFactory.getMessageQueueProducerService("redis")
                .produceMessage(new MessageQueueDto(accessLogTopic, JSONObject.toJSONString(logDto)));
    }
}

现在主体已经实现了，那redis,kafk,rabbitmq怎么设计呢 redis消息队列生产者:

/**
 * redis 消息队列
 *
 * @author starmark
 * @date 2020/5/1  上午10:41
 */
@Service
public class MessageQueueRedisProducerServiceImpl implements IMessageQueueProducerService {

    @Autowired
    private StringRedisTemplate  redisTemplate;


    @Override
    public boolean produceMessage(MessageQueueDto messageQueueDto) {
        redisTemplate.convertAndSend(messageQueueDto.getTopic(),messageQueueDto.getMessage());
        return true;
    }

    @Override
    public boolean support(String producerType) {
        return Objects.equals(producerType,"redis");
    }
}

kafka消息队列生产者:

/**
 * redis 消息队列
 *
 * @author starmark
 * @date 2020/5/1  上午10:41
 */
@Service
public class MessageQueueKafkaProducerServiceImpl implements IMessageQueueProducerService {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Override
    public boolean produceMessage(MessageQueueDto messageQueueDto) {
        kafkaTemplate.send(messageQueueDto.getTopic(),messageQueueDto.getMessage());
        return true;
    }

    @Override
    public boolean support(String producerType) {
        return Objects.equals(producerType, "kafka");
    }
}

rabbitmq生产者这个有点折腾，主要是我希望自动创建队列，但实现用的时候，要先手动创建，所以我自己想了个办法，再发消息时，判断有没有创建queue,没有的话，先创建:

/**
 * rabbitmq 消息队列
 *
 * @author starmark
 * @date 2020/5/10  上午10:41
 */
@Service
public class MessageQueueRabbitmqProducerServiceImpl implements IMessageQueueProducerService {
    private final ConfigurableApplicationContext applicationContext;

    private final List<IMessageQueueProducerService> messageQueueProducerServices;

    @Autowired
    public MessageQueueRabbitmqProducerServiceImpl(List<IMessageQueueProducerService> messageQueueProducerServiceList, ConfigurableApplicationContext applicationContext) {
        messageQueueProducerServices = messageQueueProducerServiceList.stream().filter(messageQueueConsumerService ->
                messageQueueConsumerService.support("rabbitmq")).collect(Collectors.toList());
        this.applicationContext = applicationContext;
    }

    //自动注入RabbitTemplate模板类
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Override
    public boolean produceMessage(MessageQueueDto messageQueueDto) {
       this.createQueue(messageQueueDto.getTopic());
        rabbitTemplate.convertAndSend(messageQueueDto.getTopic(), messageQueueDto.getMessage());

        return true;
    }

    @Override
    public boolean support(String producerType) {
        return Objects.equals(producerType, "rabbitmq");
    }

    /**
     * 创建rabbitmq 队例
     * @param topic　主题
     */
    private void createQueue(String topic) {
        if (!applicationContext.containsBean(topic)) {
            this.registerBean(topic  , topic);
        }
    }

    private void registerBean(String name, Object... args) {
        BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder.genericBeanDefinition(Queue.class);
        if (args.length > 0) {
            for (Object arg : args) {
                beanDefinitionBuilder.addConstructorArgValue(arg);
            }
        }
        BeanDefinition beanDefinition = beanDefinitionBuilder.getRawBeanDefinition();

        BeanDefinitionRegistry beanFactory = (BeanDefinitionRegistry) applicationContext.getBeanFactory();
        beanFactory.registerBeanDefinition(name, beanDefinition);

    }
}

至此，通用的消息队列(redis,kafka,rabbitmq)已完成，把redis,kafka,rabbitmq,的实现打包成不同的jar包，想用哪一个就用哪一个。生产者做得有点长，消费者的设计开新章吧！

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

如有侵权请联系 cloudcommunity@tencent.com 删除

rabbitmq