企业官网建设流程全解析

Spring3.5.7 RabbitMq3.13-management

在大量消息的消费阶段，涉及到数据库CURD操作时，如何去优化消费呢？

docker容器准备

docker run -d \
--name rabbitmq \
-p 5672:5672 \
-p 15672:15672 \
-e RABBITMQ_DEFAULT_USER=admin \
-e RABBITMQ_DEFAULT_PASS=admin123 \
rabbitmq:3.13-management

pom.xml

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>

application.yml

publisher

spring: datasource: driver-class-name: com.mysql.cj.jdbc.Driver url: jdbc:mysql://localhost:3306/demo?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai username: root password: 123456 application: name: publisher rabbitmq: host: port: 5672 username: admin password: virtual-host: /dlx # 虚拟主机 connection-timeout: 5s # 超时时间 template: retry: enabled: true multiplier: 1 # 再次的等待时长倍数 max-attempts: 3 # 最大重试次数 publisher-returns: false # 开启return机制,返回路由消息，一般不用管 publisher-confirm-type: none # MQ确认消息是否成功到达队列，simple同步阻塞等待回执消息，none关闭消息确认，需要在发送消息时绑定

consumer

spring: application: name: consumer rabbitmq: host: port: 5672 username: admin password: virtual-host: /dlx listener: simple: prefetch: 1 acknowledge-mode: auto retry: enabled: true initial-interval: 1000ms multiplier: 1 max-attempts: 3 stateless: true

创建两个基础的SpringBoot Module

使用的是注解形式，定义交换机，routingKey为在消费端配置真正的交换机与队列的映射关系，此处的routingKey是个代号

publisher:

@RestController @RequestMapping("/user/test") public class UserBatchTestController { private static final Logger log = LoggerFactory.getLogger(UserBatchTestController.class); private static final String EXCHANGE = "batch.user.exchange"; private static final String EXCHANGESINGLE = "single.user.exchange"; private static final String ROUTING_KEY = "batch.user"; private static final String ROUTING_KEY_SINGLE = "single.user"; private static final String[] NAMES = { "张三", "李四", "王五", "赵六", "孙七", "周八", "吴九", "郑十", "钱十一", "陈十二" }; @Resource private RabbitTemplate rabbitTemplate; @GetMapping("/sendBatch") public String sendMessagesBatch(int count) { sendMessage(count, EXCHANGE, ROUTING_KEY); return "success"; } @GetMapping("/sendSingle") public String sendMessagesSingle(int count) { sendMessage(count, EXCHANGESINGLE, ROUTING_KEY_SINGLE); return "success"; } private void sendMessage(int count, String exchange, String routingKey) { for (int i = 0; i < count; i++) { User user = new User(); user.setId((long) (i + 1)); user.setUsername(NAMES[i % NAMES.length] + "_" + (i + 1)); rabbitTemplate.convertAndSend(exchange, routingKey, user); } } }

consumer

@Configuration public class BatchConsumerConfig { @Bean("batchContainerFactory") public SimpleRabbitListenerContainerFactory batchContainerFactory( ConnectionFactory connectionFactory, Jackson2JsonMessageConverter converter) { SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setMessageConverter(converter); // false（默认）时，每拉到一条消息立刻调一次 handleBatchUsers(User user)；设 true 后变成调 handleBatchUsers(List<User> users)，框架帮你攒批再投递。 factory.setBatchListener(true); // 控制每批最多攒多少条。攒够 10 条立刻调用监听方法，不等。比如发了 25 条消息，你会收到 3 次回调：10、10、5。 factory.setBatchSize(10); // 兜底参数。**如果一直凑不够 10 条怎么办？**5 秒后不等了，有多少给多少。避免消息量小的时候第一条消息来了，第二条一直不来，第一条永远不被消费。 factory.setReceiveTimeout(5000L); // 告诉底层 AMQP 通道一次可以接收多条消息。不开启的话，即使上面写了 batchListener=true，通道也只会一次给一条，批不起来。它俩必须配合。 factory.setConsumerBatchEnabled(true); // 决定一次从 broker 预取多少条消息到本地内存。这是批量消费能成的底层前提——如果 prefetch=1，每拿到 1 条才去要下一条，batch 基本组不起来。 factory.setPrefetchCount(50); return factory; } }

各自的启动类需要配置一个

@Bean public Jackson2JsonMessageConverter jackson2JsonMessageConverter() { return new Jackson2JsonMessageConverter(); }

消费消息的注意点（待补充）

批量化的操作，索引如何取舍？

对于一张表写多读少，就舍弃一部分的读性能，减少对应的索引的构建，保留主键索引或有序性的索引

如果写多读多，考虑读写分离，分别构建索引