关于springboot:SpringBoot集成消息队列

背景

最近在对公司开发框架进行优化，框架内波及到多处入库的日志记录，例如登录日志 / 操作日志 / 拜访日志 / 业务执行日志，集成在业务代码中耦合度较高且占用业务操作执行工夫，所以筹备集成相干音讯队列进行代码解耦

计划布局

现有的成熟音讯队列组件十分多，例如 RabbitMQ，ActiveMQ，Kafka 等，思考到业务并发量不高且框架曾经利用于多个我的项目安稳运行，筹备提供基于 Redis 的音讯队列和集成 ActiveMQ 两种计划，Redis 音讯队列的益处是无需额定装置部署存量我的项目可安稳适度但音讯无奈长久化可能失落，ActiveMQ 解决方案成熟能够保障音讯长久化然而须要施行人员额定把握操作部署

对立设计

减少自定义配置指定音讯队列形式

system:
  #音讯队列形式 redis/activemq
  messageChannel: redis

定义音讯传输对立模型

public class MessageModel {
    private Class<? extends IMessageReceiver> handleClazz;
    private String bodyContent;
    private Class bodyClass;
    private HashMap extraParam;

    public MessageModel(){extraParam = new HashMap();
    }

    public Class<? extends IMessageReceiver> getHandleClazz() {return handleClazz;}

    public void setHandleClazz(Class<? extends IMessageReceiver> handleClazz) {this.handleClazz = handleClazz;}

    public HashMap getExtraParam() {return extraParam;}

    public void setExtraParam(HashMap extraParam) {this.extraParam = extraParam;}

    public String getBodyContent() {return bodyContent;}

    public void setBodyContent(String bodyContent) {this.bodyContent = bodyContent;}

    public Class getBodyClass() {return bodyClass;}

    public void setBodyClass(Class bodyClass) {this.bodyClass = bodyClass;}
}

定义规范音讯解决接口

public interface IMessageReceiver {void handleMessage(Object bodyObject, HashMap extraParam);
}

定义对立对外发送音讯工具类

@Component
public class MessageUtil {
    @Autowired
    private SystemConfig systemConfig;
    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;

    public void sendMessage(Object messageBody, Class<? extends IMessageReceiver> handleClass, HashMap<String,Object> extraParam) {MessageModel messageModel = new MessageModel();
        messageModel.setHandleClazz(handleClass);
        messageModel.setBodyClass(messageBody.getClass());
        messageModel.setBodyContent(JSON.toJSONString(messageBody));
        if (extraParam != null) {for (String key:extraParam.keySet()) {messageModel.getExtraParam().put(key,extraParam.get(key));
            }
        }
        if(systemConfig.getMessageChannel().equals("redis")){redisUtil.sendMessage("message", JSON.toJSON(messageModel));
        }else{jmsMessagingTemplate.convertAndSend("message",JSON.toJSONString(messageModel));
        }
    }
}

集成 Redis 音讯队列

pom 配置

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
            <version>2.0.1.RELEASE</version>
        </dependency>

连贯配置

spring:
  redis:
    host: localhost
    port: 6379
    password:

操作工具类

    @Autowired
    private RedisTemplate redisTemplate;
    public void sendMessage(String channel, Object message) {redisTemplate.convertAndSend(channel, message);
    }

音讯解决

@Component
@ConditionalOnProperty(name = "system.messageChannel", havingValue = "redis", matchIfMissing = true)
public class RedisMessageReceiver {public void receiveMessage(String message) {MessageModel messageModel = JSON.parseObject(message, MessageModel.class);
        IMessageReceiver receiver = SpringBootBeanUtil.getBean(messageModel.getHandleClazz());
        receiver.handleMessage(JSON.parseObject(messageModel.getBodyContent(), messageModel.getBodyClass()), messageModel.getExtraParam());
    }
}

配置注册

@Configuration
public class MessageCenter {
    @Bean
    @ConditionalOnProperty(name = "system.messageChannel", havingValue = "redis", matchIfMissing = true)
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        // 能够增加多个 messageListener，配置不同的交换机
        container.addMessageListener(listenerAdapter, new PatternTopic("message"));
        return container;
    }

    /**
     * 音讯监听器适配器，绑定音讯处理器，利用反射技术调用音讯处理器的业务办法
     *
     * @param receiver
     * @return
     */
    @Bean
    @ConditionalOnProperty(name = "system.messageChannel", havingValue = "redis", matchIfMissing = true)
    MessageListenerAdapter listenerAdapter(RedisMessageReceiver receiver) {return new MessageListenerAdapter(receiver, "receiveMessage");
    }
}

集成 ActiveMQ 音讯队列

pom 配置

        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-pool</artifactId>
            <version>5.15.0</version>
        </dependency>

留神：jdk1.8 对应版本 5.15.0

连贯配置

spring:
  activemq:
    broker-url: tcp://127.0.0.1:61616 #MQ 服务器地址
    user: admin
    password: admin
    pool:
      enabled: true

音讯解决

@Component
@ConditionalOnProperty(name = "system.messageChannel", havingValue = "activemq", matchIfMissing = false)
public class ActiveMQMessageReceiver {@JmsListener(destination = "message", containerFactory = "customQueueListener")
    public void handleMessage(String message) {MessageModel messageModel = JSON.parseObject(message, MessageModel.class);
        IMessageReceiver receiver = SpringBootBeanUtil.getBean(messageModel.getHandleClazz());
        receiver.handleMessage(JSON.parseObject(messageModel.getBodyContent(), messageModel.getBodyClass()), messageModel.getExtraParam());
    }
}

配置注册

@Configuration
@EnableJms
public class MessageCenter {@Bean(name = "customQueueListener")
    @ConditionalOnProperty(name = "system.messageChannel", havingValue = "activemq", matchIfMissing = false)
    public JmsListenerContainerFactory<?> customQueueListener(ConnectionFactory connectionFactory) {DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setPubSubDomain(false);
        factory.setConnectionFactory(connectionFactory);
        // 重连间隔时间
        factory.setRecoveryInterval(1000L);
        factory.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
        // 连接数
        factory.setConcurrency("5-10");
        // 指定工作线程池
        factory.setTaskExecutor(new ThreadPoolExecutor(5, 10, 1, TimeUnit.MINUTES,
                new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.CallerRunsPolicy()));
        return factory;
    }
}

应用示例

音讯解决

@Service
public class RequestLogMessageReceiver implements IMessageReceiver{
    @Autowired
    private F_RequestLogService requestLogService;

    @Override
    public void handleMessage(Object bodyObject, HashMap extraParam) {F_RequestLogDO requestLogDO = (F_RequestLogDO)bodyObject;
        requestLogService.insert(requestLogDO);
    }
}

发送音讯

@AutoWired
private MessageUtil messageUtil;
messageUtil.sendMessage(requestLogDO,RequestLogMessageReceiver.class,null);