接下来,我们就改造项目,实现搜索服务、商品静态页的数据同步。
发送方:商品微服务
接收方:搜索微服务、静态页微服务
接收消息后如何处理?
我们先在商品微服务leyou-item-service
中实现发送消息。
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
我们在application.yml中添加一些有关RabbitMQ的配置:
spring:
rabbitmq:
host: 192.168.56.101
username: leyou
password: leyou
virtual-host: /leyou
template:
exchange: leyou.item.exchange
publisher-confirms: true
AmqpTemplate
的配置 在GoodsService中封装一个发送消息到mq的方法:
private void sendMessage(Long id, String type){
// 发送消息
try {
this.amqpTemplate.convertAndSend("item." + type, id);
} catch (Exception e) {
logger.error("{}商品消息发送异常,商品id:{}", type, id, e);
}
}
这里没有指定交换机,因此默认发送到了配置中的:leyou.item.exchange
注意:这里要把所有异常都try起来,不能让消息的发送影响到正常的业务逻辑
然后在新增的时候调用:
修改的时候调用:
搜索服务接收到消息后要做的事情:
因为索引库的新增和修改方法是合二为一的,因此我们可以将这两类消息一同处理,删除另外处理。
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
spring:
rabbitmq:
host: 192.168.56.101
username: leyou
password: leyou
virtual-host: /leyou
这里只是接收消息而不发送,所以不用配置template相关内容。
代码:
@Component
public class GoodsListener {
@Autowired
private SearchService searchService;
/**
* 处理insert和update的消息
*
* @param id
* @throws Exception
*/
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "leyou.create.index.queue", durable = "true"),
exchange = @Exchange(
value = "leyou.item.exchange",
ignoreDeclarationExceptions = "true",
type = ExchangeTypes.TOPIC),
key = {"item.insert", "item.update"}))
public void listenCreate(Long id) throws Exception {
if (id == null) {
return;
}
// 创建或更新索引
this.searchService.createIndex(id);
}
/**
* 处理delete的消息
*
* @param id
*/
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "leyou.delete.index.queue", durable = "true"),
exchange = @Exchange(
value = "leyou.item.exchange",
ignoreDeclarationExceptions = "true",
type = ExchangeTypes.TOPIC),
key = "item.delete"))
public void listenDelete(Long id) {
if (id == null) {
return;
}
// 删除索引
this.searchService.deleteIndex(id);
}
}
这里因为要创建和删除索引,我们需要在SearchService中拓展两个方法,创建和删除索引:
public void createIndex(Long id) throws IOException {
Spu spu = this.goodsClient.querySpuById(id);
// 构建商品
Goods goods = this.buildGoods(spu);
// 保存数据到索引库
this.goodsRepository.save(goods);
}
public void deleteIndex(Long id) {
this.goodsRepository.deleteById(id);
}
创建索引的方法可以从之前导入数据的测试类中拷贝和改造。
商品静态页服务接收到消息后的处理:
不过,我们编写的创建静态页的方法也具备覆盖以前页面的功能,因此:增和改的消息可以放在一个方法中处理,删除消息放在另一个方法处理。
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
spring:
rabbitmq:
host: 192.168.56.101
username: leyou
password: leyou
virtual-host: /leyou
这里只是接收消息而不发送,所以不用配置template相关内容。
代码:
@Component
public class GoodsListener {
@Autowired
private GoodsHtmlService goodsHtmlService;
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "leyou.create.web.queue", durable = "true"),
exchange = @Exchange(
value = "leyou.item.exchange",
ignoreDeclarationExceptions = "true",
type = ExchangeTypes.TOPIC),
key = {"item.insert", "item.update"}))
public void listenCreate(Long id) throws Exception {
if (id == null) {
return;
}
// 创建页面
goodsHtmlService.createHtml(id);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "leyou.delete.web.queue", durable = "true"),
exchange = @Exchange(
value = "leyou.item.exchange",
ignoreDeclarationExceptions = "true",
type = ExchangeTypes.TOPIC),
key = "item.delete"))
public void listenDelete(Long id) {
if (id == null) {
return;
}
// 创建页面
goodsHtmlService.deleteHtml(id);
}
}
public void deleteHtml(Long id) {
File file = new File("C:\\project\\nginx-1.14.0\\html\\item\\", id + ".html");
file.deleteOnExit();
}
重新启动项目,并且登录RabbitMQ管理界面:http://192.168.56.101:15672
可以看到,交换机已经创建出来了:
队列也已经创建完毕:
并且队列都已经绑定到交换机:
在后台修改商品数据的价格,分别在搜索及商品详情页查看是否统一。