专栏首页JAVA后端开发通用的消息队列(redis,kafka,rabbitmq)--生产者篇

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

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

  1. 生产者通用消息对象,里面只有主题及消息
@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包,想用哪一个就用哪一个。 生产者做得有点长,消费者的设计开新章吧!

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • activiti 自定义函数解析juel表达式

    activiti是支持juel表达式的,这个也很好用,但实际过程中需要支持类方法及变量。    原来项目中解析juel,这边有自定义一个方法的 代码如下:

    星痕
  • flowable实现流程全局事件

    最近在研究flowable,发现这个东东虽说是activiti的升级版,但感觉还是没有a5的好用。 项目中需要实现一个全局事件,实现如下:

    星痕
  • spring secutiry密码验证的另一种解决办法(荐)

    最近在集成sping security 到spring boot中,看到大部分登陆验证的代码都这么写

    星痕
  • JVM应用度量框架Micrometer实战

    最近线上的项目使用了spring-actuator做度量统计收集,使用Prometheus进行数据收集,Grafana进行数据展示,用于监控生成环境机器的性能指...

    Throwable
  • SpringBoot的埋点监控你做了吗

    spring-actuator做度量统计收集,使用Prometheus(普罗米修斯)进行数据收集,Grafana(增强ui)进行数据展示,用于监控生成环境机器的...

    Java小咖秀
  • 给你的SpringBoot做埋点监控--JVM应用度量框架Micrometer

    spring-actuator做度量统计收集,使用Prometheus(普罗米修斯)进行数据收集,Grafana(增强ui)进行数据展示,用于监控生成环境机器的...

    yunlgonn
  • 突破Java面试(42) - Redis & ZooKeeper两种分布式锁实现的优劣

    一般实现分布式锁都有哪些方式?使用redis如何设计分布式锁?使用zk来设计分布式锁可以吗?这两种分布式锁的实现方式哪种效率比较高?

    JavaEdge
  • C#编写影院售票系统(A project with a higher amount of gold )(2:相关代码)

    此篇文章为项目代码,,,需要项目需求 ,思路分析与窗体效果请访问:http://www.cnblogs.com/lsy131479/p/8367304.html...

    房上的猫
  • Flex 调用添加了SoapHeader的web service

    using System.Web.Services; using System.Web.Services.Protocols;

    用户3135539
  • 学界 | CVPR 2018接收论文公布,上海交通大学6篇论文简介

    机器之心报道 作者:吴欣 不久之前,CVPR 2018 论文接收列表公布。据机器之心了解,上海交通大学电子系人工智能实验室倪冰冰教授课题组有 6 篇论文入选,本...

    机器之心

扫码关注云+社区

领取腾讯云代金券