分布式锁&kafka事务提交等编码技巧

青旅半醒 2023-09-27 08:12 17阅读 0赞

## 一、前言 ##

在开发过程中，遇到了一些比较实用的编码技巧，故记录以加深印象。因为每个技巧的篇幅较短，故不做拆分，只合在一篇小文章中阐述。以下会涉及kafka的事务提交方法、redis分布式锁简化以及多key情况下应该怎么加锁、业务日志如何解耦。

####  ####

## 二、kafka的事务提交方法 ##

kafka我们常用于削峰填谷，以及系统间解耦等场景。这里我们会常遇到一种情况，就是上游系统在处理完成业务后，需要通知其它的系统，假如我们不考虑事务提交失败的情况下，就可以像下面这样写。但是假如出现网络异常或者数据库异常等情况，就会出现事务提交失败从而回滚，但是消息却已经发生给其它服务了，那么就会导致整条调用链的异常

@Autowired      
     private KafkaTemplate kafkaTemplate;
    
     @Transactional(rollbackFor = Exception.class)
     public void saveServiceOrder(ServiceOrder serviceOrder){
         // do something
    
         NoticeListDTO notice =  NoticeListDTO.builder().build();
         // 通知其它服务
         kafkaTemplate.send(TopicNameConstants.SERVICE_ORDER_CHANGE_NOTIFY, JSONObject.toJSONString(notice));
     }

所以，我们可以进行进一步优化，就是将消息通知后置到事务提交后，这样系统的可靠度就会更高。我们增加一个kafka帮助类，如下：

@Component
     @Slf4j
     public class KafkaTemplateHelper {
         @Autowired
         private KafkaTemplate kafkaTemplate;
    
         /**
          * 事务提交后发送kafka消息
          * @param topic
          * @param data
          * @param <T>
          */
         public <T> void send(String topic, Object data) {
             // 是否开启事务判断
             if (TransactionSynchronizationManager.isSynchronizationActive()) {
                 TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
                     @Override
                     public void afterCommit() {
                         log.info("事务提交成功后发送消息:topic:{},data:{}",topic, JSONObject.toJSONString(data));
                         kafkaTemplate.send(topic,data);
                     }
                 });
             } else {
                 log.info("没有开启事务直接发送消息:topic:{},data:{}",topic, JSONObject.toJSONString(data));
                 kafkaTemplate.send(topic,data);
             }
         }
     }

kafka调用如下，它就会保证在事务结束后再通知其它系统，同理，很多需要后置的操作也可以这么玩。其实kafka还有一套可靠性应用方案可以分享，待有空再写

@Autowired
     private KafkaTemplateHelper kafkaTemplateHelper;
    
     @Transactional(rollbackFor = Exception.class)
     public void saveServiceOrder(ServiceOrder serviceOrder){
         // do something
    
         NoticeListDTO notice =  NoticeListDTO.builder().build();
         // 通知a服务
         kafkaTemplateHelper.send(TopicNameConstants.SERVICE_ORDER_CHANGE_NOTIFY, JSONObject.toJSONString(notice));
     }

####  ####

## 三、redis分布式锁代码简化 ##

我们使用redis分布式锁就离不开redission组件，举个栗子，我们一般在服务集群的情况下，为了保证并发不出现问题，会如下加锁，用一段字符串加上入参中的唯一编号(如用户id、订单编号等等)来保证接口幂等性（PS:redissonDistributedLocker只是redission的简单封装）。这样写很好，没有问题，但是我们不禁会想，好像每个创建、更新等业务操作都得给接口加这些重复代码，那么有没有更加优雅的方式呢，没错，