何冠元 / canal-mq-client · 提交

转到一个项目

作者 [wgf]
提交 15c5318013411f79df2bbc0086173d1a500ca79a 1 个父辈 2c602c40

亚马逊订单数据同步（非主键）

内嵌 并排对比
正在显示 11 个修改的文件 包含 125 行增加 和 92 行删除
@@ -30,6 +30,26 @@ @@ -30,6 +30,26 @@
30 </dependency> 30 </dependency>
31 31
32 <dependency> 32 <dependency>
  33 + <groupId>org.springframework.boot</groupId>
  34 + <artifactId>spring-boot-starter-log4j2</artifactId>
  35 + </dependency>
  36 +
  37 + <dependency>
  38 + <groupId>org.springframework.boot</groupId>
  39 + <artifactId>spring-boot-starter</artifactId>
  40 + <exclusions>
  41 + <exclusion>
  42 + <artifactId>logback-classic</artifactId>
  43 + <groupId>ch.qos.logback</groupId>
  44 + </exclusion>
  45 + <exclusion>
  46 + <artifactId>spring-boot-starter-logging</artifactId>
  47 + <groupId>org.springframework.boot</groupId>
  48 + </exclusion>
  49 + </exclusions>
  50 + </dependency>
  51 +
  52 + <dependency>
33 <groupId>com.alibaba</groupId> 53 <groupId>com.alibaba</groupId>
34 <artifactId>fastjson</artifactId> 54 <artifactId>fastjson</artifactId>
35 <version>1.2.73</version> 55 <version>1.2.73</version>
@@ -28,7 +28,7 @@ public class AmazonOrderItemListener { @@ -28,7 +28,7 @@ public class AmazonOrderItemListener {
28 @Autowired 28 @Autowired
29 AmazonOrderItemService amazonOrderItemService; 29 AmazonOrderItemService amazonOrderItemService;
30 30
31 - @RabbitListener(queues = "polaris_order_center.amazon_order_item") 31 + @RabbitListener(queues = "polaris_order_center.amazon_order_item.by")
32 public void receive(String message, Channel channel, Message messageEntity) throws IOException { 32 public void receive(String message, Channel channel, Message messageEntity) throws IOException {
33 33
34 try { 34 try {
@@ -38,8 +38,6 @@ public class AmazonOrderItemListener { @@ -38,8 +38,6 @@ public class AmazonOrderItemListener {
38 38
39 EventType eventType = EventType.valueOf(messageVo.getType()); 39 EventType eventType = EventType.valueOf(messageVo.getType());
40 40
41 - log.info("当前监听binlog 数据库：{}, 数据表：{}, 数据：{}", messageVo.getDatabase(), messageVo.getTable(), message);  
42 -  
43 // 数据同步只关注这三种事件 41 // 数据同步只关注这三种事件
44 switch (eventType) { 42 switch (eventType) {
45 case INSERT: 43 case INSERT:
@@ -3,7 +3,6 @@ package com.aukey.example.listener; @@ -3,7 +3,6 @@ package com.aukey.example.listener;
3 import com.alibaba.fastjson.JSON; 3 import com.alibaba.fastjson.JSON;
4 import com.alibaba.fastjson.TypeReference; 4 import com.alibaba.fastjson.TypeReference;
5 import com.aukey.example.constant.EventType; 5 import com.aukey.example.constant.EventType;
6 -import com.aukey.example.constant.MQConst;  
7 import com.aukey.example.entity.fbaStock.AmazonOrder; 6 import com.aukey.example.entity.fbaStock.AmazonOrder;
8 import com.aukey.example.service.fbaStock.AmazonOrderService; 7 import com.aukey.example.service.fbaStock.AmazonOrderService;
9 import com.aukey.example.vo.MessageVo; 8 import com.aukey.example.vo.MessageVo;
@@ -28,7 +27,7 @@ public class AmazonOrderListener { @@ -28,7 +27,7 @@ public class AmazonOrderListener {
28 @Autowired 27 @Autowired
29 AmazonOrderService amazonOrderService; 28 AmazonOrderService amazonOrderService;
30 29
31 - @RabbitListener(queues = "polaris_order_center.amazon_order") 30 + @RabbitListener(queues = "polaris_order_center.amazon_order.by")
32 public void receive(String message, Channel channel, Message messageEntity) throws IOException { 31 public void receive(String message, Channel channel, Message messageEntity) throws IOException {
33 32
34 try { 33 try {
@@ -38,8 +37,6 @@ public class AmazonOrderListener { @@ -38,8 +37,6 @@ public class AmazonOrderListener {
38 37
39 EventType eventType = EventType.valueOf(messageVo.getType()); 38 EventType eventType = EventType.valueOf(messageVo.getType());
40 39
41 - log.info("当前监听binlog 数据库：{}, 数据表：{}, 数据：{}", messageVo.getDatabase(), messageVo.getTable(), message);  
42 -  
43 // 数据同步只关注这三种事件 40 // 数据同步只关注这三种事件
44 switch (eventType) { 41 switch (eventType) {
45 case INSERT: 42 case INSERT:
1 -package com.aukey.example.listener;  
2 -  
3 -import com.alibaba.fastjson.JSON;  
4 -import com.alibaba.fastjson.TypeReference;  
5 -import com.aukey.example.constant.EventType;  
6 -import com.aukey.example.constant.MQConst;  
7 -import com.aukey.example.entity.AmazonOrder;  
8 -import com.aukey.example.vo.MessageVo;  
9 -import com.rabbitmq.client.Channel;  
10 -import lombok.extern.slf4j.Slf4j;  
11 -import org.springframework.amqp.core.Message;  
12 -import org.springframework.amqp.rabbit.annotation.RabbitListener;  
13 -import org.springframework.stereotype.Component;  
14 -  
15 -import java.io.IOException;  
16 -  
17 -/**  
18 - * @author: wgf  
19 - * @create: 2020-06-09 14:30  
20 - * @description: CanalUser监听器  
21 - **/  
22 -@Slf4j  
23 -@Component  
24 -public class CanalUserListener {  
25 -  
26 - @RabbitListener(queues = "polaris_order_center.amazon_order")  
27 - public void receive(String message, Channel channel, Message messageEntity) throws IOException {  
28 -  
29 - try {  
30 -  
31 - log.info("接收到队列: {} 消息：{}", MQConst.TEST, message);  
32 -  
33 - MessageVo<AmazonOrder> messageVo = JSON.parseObject(message, new TypeReference<MessageVo<AmazonOrder>>() {  
34 - });  
35 -  
36 - EventType eventType = EventType.valueOf(messageVo.getType());  
37 -  
38 - log.info("当前监听binlog 数据库：{}, 数据表：{}", messageVo.getDatabase(), messageVo.getTable());  
39 -  
40 - // 数据同步只关注这三种事件  
41 - switch (eventType) {  
42 - case INSERT:  
43 - log.info("触发 INSERT 事件");  
44 - // messageVo.getData(); 获取数据变更  
45 - // TODO 自定义实现  
46 - break;  
47 - case UPDATE:  
48 - log.info("触发 UPDATE 事件");  
49 - // TODO 自定义实现  
50 - break;  
51 - case DELETE:  
52 - log.info("触发 DELETE 事件");  
53 - // TODO 自定义实现  
54 - break;  
55 - default:  
56 - log.info("其他事件类型：{}, 过滤不处理", eventType);  
57 - }  
58 -  
59 - log.info("消息长度：{}", messageVo.getData().size());  
60 -  
61 - /**  
62 - * 消息消费确认  
63 - * 如果客户端在线没有签收没有签收这条Message，则此消息进入Unacked状态，此时监听器阻塞等待消息确认，不推送新Message  
64 - * 如果待消息确认并且客户端下线，下次客户端上线重新推送上次Unacked状态Message  
65 - */  
66 - channel.basicAck(messageEntity.getMessageProperties().getDeliveryTag(), false);  
67 - } catch (Exception e) {  
68 - /**  
69 - * 第一个参数deliveryTag：发布的每一条消息都会获得一个唯一的deliveryTag，deliveryTag在channel范围内是唯一的  
70 - * 第二个参数multiple：批量确认标志。如果值为true，包含本条消息在内的、所有比该消息deliveryTag值小的 消息都被拒绝了（除了已经被 ack 的以外）;如果值为false，只拒绝三本条消息  
71 - * 第三个参数requeue：表示如何处理这条消息，如果值为true，则重新放入RabbitMQ的发送队列，如果值为false，则通知RabbitMQ销毁这条消息  
72 - */  
73 - //channel.basicNack(messageEntity.getMessageProperties().getDeliveryTag(), false,true);  
74 - e.printStackTrace();  
75 - }  
76 - }  
77 -}  
@@ -4,4 +4,7 @@ import com.aukey.example.commom.mapper.Mapper; @@ -4,4 +4,7 @@ import com.aukey.example.commom.mapper.Mapper;
4 import com.aukey.example.entity.fbaStock.AmazonOrderItem; 4 import com.aukey.example.entity.fbaStock.AmazonOrderItem;
5 5
6 public interface AmazonOrderItemMapper extends Mapper<AmazonOrderItem> { 6 public interface AmazonOrderItemMapper extends Mapper<AmazonOrderItem> {
  7 +
  8 + Integer selectAid(AmazonOrderItem amazonOrderItem);
  9 +
7 } 10 }
@@ -4,4 +4,5 @@ import com.aukey.example.commom.mapper.Mapper; @@ -4,4 +4,5 @@ import com.aukey.example.commom.mapper.Mapper;
4 import com.aukey.example.entity.fbaStock.AmazonOrder; 4 import com.aukey.example.entity.fbaStock.AmazonOrder;
5 5
6 public interface AmazonOrderMapper extends Mapper<AmazonOrder> { 6 public interface AmazonOrderMapper extends Mapper<AmazonOrder> {
  7 + Integer selectAid(AmazonOrder amazonOrder);
7 } 8 }
@@ -3,9 +3,11 @@ package com.aukey.example.service.fbaStock.impl; @@ -3,9 +3,11 @@ package com.aukey.example.service.fbaStock.impl;
3 import com.aukey.example.entity.fbaStock.AmazonOrderItem; 3 import com.aukey.example.entity.fbaStock.AmazonOrderItem;
4 import com.aukey.example.mapper.fbaStock.AmazonOrderItemMapper; 4 import com.aukey.example.mapper.fbaStock.AmazonOrderItemMapper;
5 import com.aukey.example.service.fbaStock.AmazonOrderItemService; 5 import com.aukey.example.service.fbaStock.AmazonOrderItemService;
  6 +import lombok.extern.slf4j.Slf4j;
6 import org.springframework.beans.factory.annotation.Autowired; 7 import org.springframework.beans.factory.annotation.Autowired;
7 import org.springframework.stereotype.Service; 8 import org.springframework.stereotype.Service;
8 import org.springframework.transaction.annotation.Transactional; 9 import org.springframework.transaction.annotation.Transactional;
  10 +import tk.mybatis.mapper.entity.Example;
9 11
10 import java.util.List; 12 import java.util.List;
11 import java.util.Objects; 13 import java.util.Objects;
@@ -15,6 +17,7 @@ import java.util.Objects; @@ -15,6 +17,7 @@ import java.util.Objects;
15 * @create: 2020-08-25 14:36 17 * @create: 2020-08-25 14:36
16 * @description: 18 * @description:
17 **/ 19 **/
  20 +@Slf4j
18 @Service 21 @Service
19 public class AmazonOrderItemServiceImpl implements AmazonOrderItemService { 22 public class AmazonOrderItemServiceImpl implements AmazonOrderItemService {
20 23
@@ -25,19 +28,22 @@ public class AmazonOrderItemServiceImpl implements AmazonOrderItemService { @@ -25,19 +28,22 @@ public class AmazonOrderItemServiceImpl implements AmazonOrderItemService {
25 @Transactional(value = "fbaStockTransactionManager") 28 @Transactional(value = "fbaStockTransactionManager")
26 public void insertOrUpdate(List<AmazonOrderItem> list) { 29 public void insertOrUpdate(List<AmazonOrderItem> list) {
27 for (AmazonOrderItem amazonOrderItem : list) { 30 for (AmazonOrderItem amazonOrderItem : list) {
  31 + log.info("订单号：{}", amazonOrderItem.getAmazonOrderId());
28 AmazonOrderItem query = new AmazonOrderItem(); 32 AmazonOrderItem query = new AmazonOrderItem();
  33 + query.setAmazonOrderId(amazonOrderItem.getAmazonOrderId());
29 query.setAccountId(amazonOrderItem.getAccountId()); 34 query.setAccountId(amazonOrderItem.getAccountId());
30 query.setAreaId(amazonOrderItem.getAreaId()); 35 query.setAreaId(amazonOrderItem.getAreaId());
31 - query.setAmazonOrderId(amazonOrderItem.getAmazonOrderId());  
32 - AmazonOrderItem existOrder = this.amazonOrderItemMapper.selectOne(query); 36 + query.setSellerSku(amazonOrderItem.getSellerSku());
  37 + query.setOrderItemId(amazonOrderItem.getOrderItemId());
  38 + Integer aid = this.amazonOrderItemMapper.selectAid(query);
33 39
34 // 主键和 headId不能使用 40 // 主键和 headId不能使用
35 amazonOrderItem.setOrderHeaderId(null); 41 amazonOrderItem.setOrderHeaderId(null);
36 - if (Objects.isNull(existOrder)) { 42 + if (Objects.isNull(aid)) {
37 amazonOrderItem.setAid(null); 43 amazonOrderItem.setAid(null);
38 this.amazonOrderItemMapper.insertSelective(amazonOrderItem); 44 this.amazonOrderItemMapper.insertSelective(amazonOrderItem);
39 } else { 45 } else {
40 - amazonOrderItem.setAid(existOrder.getAid()); 46 + amazonOrderItem.setAid(aid);
41 this.amazonOrderItemMapper.updateByPrimaryKeySelective(amazonOrderItem); 47 this.amazonOrderItemMapper.updateByPrimaryKeySelective(amazonOrderItem);
42 } 48 }
43 } 49 }
@@ -3,9 +3,11 @@ package com.aukey.example.service.fbaStock.impl; @@ -3,9 +3,11 @@ package com.aukey.example.service.fbaStock.impl;
3 import com.aukey.example.entity.fbaStock.AmazonOrder; 3 import com.aukey.example.entity.fbaStock.AmazonOrder;
4 import com.aukey.example.mapper.fbaStock.AmazonOrderMapper; 4 import com.aukey.example.mapper.fbaStock.AmazonOrderMapper;
5 import com.aukey.example.service.fbaStock.AmazonOrderService; 5 import com.aukey.example.service.fbaStock.AmazonOrderService;
  6 +import lombok.extern.slf4j.Slf4j;
6 import org.springframework.beans.factory.annotation.Autowired; 7 import org.springframework.beans.factory.annotation.Autowired;
7 import org.springframework.stereotype.Service; 8 import org.springframework.stereotype.Service;
8 import org.springframework.transaction.annotation.Transactional; 9 import org.springframework.transaction.annotation.Transactional;
  10 +import tk.mybatis.mapper.entity.Example;
9 11
10 import java.util.List; 12 import java.util.List;
11 import java.util.Objects; 13 import java.util.Objects;
@@ -15,6 +17,7 @@ import java.util.Objects; @@ -15,6 +17,7 @@ import java.util.Objects;
15 * @create: 2020-08-25 14:36 17 * @create: 2020-08-25 14:36
16 * @description: 18 * @description:
17 **/ 19 **/
  20 +@Slf4j
18 @Service 21 @Service
19 public class AmazonOrderServiceImpl implements AmazonOrderService { 22 public class AmazonOrderServiceImpl implements AmazonOrderService {
20 23
@@ -25,18 +28,19 @@ public class AmazonOrderServiceImpl implements AmazonOrderService { @@ -25,18 +28,19 @@ public class AmazonOrderServiceImpl implements AmazonOrderService {
25 @Transactional(value = "fbaStockTransactionManager") 28 @Transactional(value = "fbaStockTransactionManager")
26 public void insertOrUpdate(List<AmazonOrder> list) { 29 public void insertOrUpdate(List<AmazonOrder> list) {
27 for (AmazonOrder amazonOrder : list) { 30 for (AmazonOrder amazonOrder : list) {
  31 + log.info("订单号：{}", amazonOrder.getAmazonOrderId());
28 AmazonOrder query = new AmazonOrder(); 32 AmazonOrder query = new AmazonOrder();
  33 + query.setAmazonOrderId(amazonOrder.getAmazonOrderId());
29 query.setAccountId(amazonOrder.getAccountId()); 34 query.setAccountId(amazonOrder.getAccountId());
30 query.setAreaId(amazonOrder.getAreaId()); 35 query.setAreaId(amazonOrder.getAreaId());
31 - query.setAmazonOrderId(amazonOrder.getAmazonOrderId());  
32 - AmazonOrder existOrder = this.amazonOrderMapper.selectOne(query); 36 + Integer aid = this.amazonOrderMapper.selectAid(query);
33 37
34 // 主键不能使用 38 // 主键不能使用
35 - if (Objects.isNull(existOrder)) { 39 + if (Objects.isNull(aid)) {
36 amazonOrder.setAid(null); 40 amazonOrder.setAid(null);
37 this.amazonOrderMapper.insertSelective(amazonOrder); 41 this.amazonOrderMapper.insertSelective(amazonOrder);
38 } else { 42 } else {
39 - amazonOrder.setAid(existOrder.getAid()); 43 + amazonOrder.setAid(aid);
40 this.amazonOrderMapper.updateByPrimaryKeySelective(amazonOrder); 44 this.amazonOrderMapper.updateByPrimaryKeySelective(amazonOrder);
41 } 45 }
42 } 46 }
  1 +<?xml version="1.0" encoding="UTF-8"?>
  2 +<Configuration status="WARN">
  3 +
  4 + <Properties>
  5 + <Property name="PID">????</Property>
  6 + <Property name="LOG_EXCEPTION_CONVERSION_WORD">%xwEx</Property>
  7 + <Property name="LOG_LEVEL_PATTERN">%5p</Property>
  8 + <Property name="LOG_PATTERN">%clr{%d{MM-dd HH:mm:ss}}{faint} %clr{${LOG_LEVEL_PATTERN}} %clr{${sys:PID}}{magenta} %clr{-}{faint} %clr{[%t]}{faint} %clr{%c{1}}{cyan} %clr{:}{faint} %m%n${sys:LOG_EXCEPTION_CONVERSION_WORD}</Property>
  9 + </Properties>
  10 +
  11 + <Appenders>
  12 + <Console name="Console" target="SYSTEM_OUT" follow="true">
  13 + <PatternLayout pattern="${LOG_PATTERN}" />
  14 + </Console>
  15 +
  16 + <RollingFile name="RollingFileInfo" fileName="logs/info.log"
  17 + filePattern="logs/$${date:yyyy-MM}/info-%d{yyyy-MM-dd}-%i.log">
  18 + <ThresholdFilter level="info" onMatch="ACCEPT" onMismatch="DENY"/>
  19 + <PatternLayout pattern="[%d{HH:mm:ss:SSS}] [%p] - %l - %m%n"/>
  20 + <Policies>
  21 + <TimeBasedTriggeringPolicy/>
  22 + <SizeBasedTriggeringPolicy size="350 MB"/>
  23 + </Policies>
  24 + </RollingFile>
  25 +
  26 + <RollingFile name="RollingFileError" fileName="logs/error.log"
  27 + filePattern="logs/$${date:yyyy-MM}/error-%d{yyyy-MM-dd}-%i.log">
  28 + <ThresholdFilter level="error" onMatch="ACCEPT" onMismatch="DENY"/>
  29 + <PatternLayout pattern="[%d{HH:mm:ss:SSS}] [%p] - %l - %m%n"/>
  30 + <Policies>
  31 + <TimeBasedTriggeringPolicy/>
  32 + <SizeBasedTriggeringPolicy size="350 MB"/>
  33 + </Policies>
  34 + </RollingFile>
  35 +
  36 + </Appenders>
  37 +
  38 + <Loggers>
  39 +
  40 + <Logger name="org.apache.catalina.startup.DigesterFactory" level="error" />
  41 + <Logger name="org.apache.catalina.util.LifecycleBase" level="error" />
  42 + <Logger name="org.apache.coyote.http11.Http11NioProtocol" level="warn" />
  43 + <logger name="org.apache.sshd.common.util.SecurityUtils" level="warn"/>
  44 + <Logger name="org.apache.tomcat.util.net.NioSelectorPool" level="warn" />
  45 + <Logger name="org.crsh.plugin" level="warn" />
  46 + <logger name="org.crsh.ssh" level="warn"/>
  47 + <Logger name="org.eclipse.jetty.util.component.AbstractLifeCycle" level="error" />
  48 + <Logger name="org.hibernate.validator.internal.util.Version" level="warn" />
  49 + <logger name="org.springframework.boot.actuate.autoconfigure.CrshAutoConfiguration" level="warn"/>
  50 + <logger name="org.springframework.boot.actuate.endpoint.jmx" level="warn"/>
  51 + <logger name="org.thymeleaf" level="warn"/>
  52 + <!-- 打印sql -->
  53 + <!--<logger name="com.aukey.example.mapper" level="DEBUG"/> -->
  54 + <Root level="INFO">
  55 + <AppenderRef ref="Console" />
  56 + <AppenderRef ref="RollingFileInfo" />
  57 + <AppenderRef ref="RollingFileError" />
  58 + </Root>
  59 +
  60 + </Loggers>
  61 +
  62 +
  63 +</Configuration>
@@ -59,4 +59,14 @@ @@ -59,4 +59,14 @@
59 <result column="title" jdbcType="LONGVARBINARY" property="titleAlias" /> 59 <result column="title" jdbcType="LONGVARBINARY" property="titleAlias" />
60 <result column="gift_message_text" jdbcType="LONGVARBINARY" property="giftMessageText" /> 60 <result column="gift_message_text" jdbcType="LONGVARBINARY" property="giftMessageText" />
61 </resultMap> 61 </resultMap>
  62 +
  63 + <select id="selectAid" resultType="java.lang.Integer">
  64 + select aid
  65 + from amazon_order_item
  66 + where amazon_order_id = #{amazonOrderId}
  67 + and account_id = #{accountId}
  68 + and area_id = #{areaId}
  69 + and seller_sku = #{sellerSku}
  70 + and order_item_id = #{orderItemId}
  71 + </select>
62 </mapper> 72 </mapper>
@@ -60,4 +60,12 @@ @@ -60,4 +60,12 @@
60 <result column="shipping_address_line2" jdbcType="BLOB" property="shippingAddressLine2" /> 60 <result column="shipping_address_line2" jdbcType="BLOB" property="shippingAddressLine2" />
61 <result column="shipping_address_line3" jdbcType="BLOB" property="shippingAddressLine3" /> 61 <result column="shipping_address_line3" jdbcType="BLOB" property="shippingAddressLine3" />
62 </resultMap> 62 </resultMap>
  63 +
  64 + <select id="selectAid" resultType="java.lang.Integer">
  65 + SELECT aid
  66 + FROM amazon_order
  67 + WHERE amazon_order_id = #{amazonOrderId}
  68 + and account_id = #{accountId}
  69 + and area_id = #{areaId}
  70 + </select>
63 </mapper> 71 </mapper>