网上有很多消息队列的中间件，如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包，想用哪一个就用哪一个。
生产者做得有点长，消费者的设计开新章吧！
