Practical Coding Tips: Kafka Transaction Commit, Redis Distributed Lock Simplification, AOP Lock Annotation, and Business Log Decoupling

During development the author records several practical coding techniques to improve reliability and maintainability, covering Kafka transaction commit handling, Redis distributed lock simplification, multi‑key locking strategies, and business log decoupling.

Kafka transaction commit method – By placing the Kafka send inside a Spring @Transactional method, failures in the database can trigger a rollback, but the message may already be sent. The author first shows a basic implementation and then introduces a helper class that sends the message only after the transaction successfully commits.

@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));
}

The improved KafkaTemplateHelper checks whether a transaction is active and registers a synchronization callback to send the message after commit, falling back to immediate send when no transaction is present.

@Component
@Slf4j
public class KafkaTemplateHelper {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 事务提交后发送kafka消息
     */
    public 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);
        }
    }
}

Usage with the helper:

@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 distributed lock simplification – The author demonstrates using Redisson to acquire a lock for a single operation, then abstracts the pattern into an AOP annotation.

@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
public @interface AopLock {
    /** SpEL expression to compute the lock key */
    String value();
    /** wait time in seconds */
    int waitTime() default 0;
    /** lease time in seconds */
    int leaseTime() default 6;
    int errorCode() default 2733;
    String errorMsg() default "操作过于频繁，请稍后再试";
}

The corresponding aspect acquires the Redisson lock before method execution and releases it afterwards, throwing a custom exception if the lock cannot be obtained.

@Slf4j
@Aspect
@Order(3)
@ConditionalOnBean(RedissonClient.class)
public class AopLockAspect {
    @Autowired
    private RedissonClient redissonClient;

    @Value("${spring.application.name}")
    private String lockKeyPrefix;

    @Around("@annotation(common.aop.annos.AopLock)")
    public Object lock(ProceedingJoinPoint joinPoint) throws Throwable {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        Method method = signature.getMethod();
        AopLock aopLock = method.getAnnotation(AopLock.class);
        String spEl = aopLock.value();
        String expressionValue = lockKeyPrefix + ":" + PARSER.parseExpression(spEl).getValue(initEvaluationContext(joinPoint));
        RLock lock = redissonClient.getLock(expressionValue);
        try {
            boolean getterLock = lock.tryLock(aopLock.waitTime(), aopLock.leaseTime(), TimeUnit.SECONDS);
            if (!getterLock) {
                throw new ServiceException(aopLock.errorCode(), aopLock.errorMsg());
            }
            return joinPoint.proceed(joinPoint.getArgs());
        } finally {
            try { lock.unlock(); } catch (Exception e) { log.warn("unlock error:" + e.getMessage(), e); }
        }
    }
}

Example of applying the annotation:

@ApiOperation("服务单更新")
@AopLock(value = "'mh:scs:serviceOrderUpdate:' + #req.serviceOrderId", leaseTime = 60 * 30)
@Transactional(rollbackFor = Exception.class)
public ApiResult serviceOrderUpdate(@RequestBody @Validated ServiceOrder req) {
    log.info("服务单更新:time={},params={}", System.currentTimeMillis(), JSONObject.toJSONString(req));
    // do something
    return ApiResult.success();
}

Multi‑key locking scenario – For batch operations where several resources need to be locked simultaneously, the author shows how to collect keys, acquire locks for each, and release them in a finally block.

@ApiOperation("批量更新服务单信息")
@PostMapping("/xxxx/updateServiceOrders")
public ResponseBean updateServiceOrders(@RequestBody @Validated UpdateServiceOrdersReq req) {
    List<String> redisKeys = new ArrayList<>();
    for (ServiceOrder serviceOrder : req.getServiceOrderList()) {
        redisKeys.add("mh:scs:updateServiceOrders:" + serviceOrder.getServiceOrderId());
    }
    try {
        for (String redisKey : redisKeys) {
            boolean lock = redissonDistributedLocker.tryLock(redisKey, 5L, 30L);
            if (!lock) {
                AssertUtil.businessInvalid("批量更新服务单获取锁失败,请稍后尝试!");
            }
        }
        // do something
        return ResponseBean.success();
    } finally {
        redisKeys.forEach(key -> {
            try { redissonDistributedLocker.unlock(key); }
            catch (Exception e) { log.error("updateServiceOrders:释放redis锁失败:{}", key, e); }
        });
    }
}

Business log decoupling – To avoid slowing down the main thread, logging is off‑loaded to a dedicated thread pool. The utility class provides methods to create log entities, batch‑insert them asynchronously, and truncate overly long content.

public class ServiceOrderLogUtils {
    private static ScsServiceOrderLogService logService = ApplicationContextUtils.getBean(ScsServiceOrderLogService.class);
    private static int corePoolSize = Runtime.getRuntime().availableProcessors();
    private static ExecutorService executor = new ThreadPoolExecutor(
        corePoolSize, corePoolSize * 2, 10L, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(Integer.MAX_VALUE), new ThreadPoolExecutor.CallerRunsPolicy());

    public static void execute(Runnable runnable) { executor.execute(runnable); }

    public static void saveLog(Long serviceOrderId, OperTypeEnum operType, String operContent) {
        saveLog(createLog(serviceOrderId, operType, operContent));
    }

    public static ScsServiceOrderLog createLog(Long serviceOrderId, OperTypeEnum operType, String operContent) {
        AuthUser userInfo = WebUtils.getCurrentUser();
        return createLog(serviceOrderId, operType, operContent, StringUtil.toInt(userInfo.getLoginName()), userInfo.getName());
    }

    public static void saveLog(ScsServiceOrderLog log) {
        List<ScsServiceOrderLog> list = new ArrayList<>();
        list.add(log);
        saveLog(list);
    }

    public static void saveLog(List<ScsServiceOrderLog> list) {
        if (CollectionUtils.isEmpty(list)) return;
        Date now = new Date();
        for (ScsServiceOrderLog log : list) {
            if (log.getOperatorTime() == null) log.setOperatorTime(now);
            if (StrUtil.length(log.getOperContent()) > 512) {
                log.setOperContent(StrUtil.subWithLength(log.getOperContent(), 0, 512));
            }
        }
        execute(new SaveLogThread(list));
    }

    static class SaveLogThread implements Runnable {
        private List<ScsServiceOrderLog> list;
        SaveLogThread(List<ScsServiceOrderLog> list) { this.list = list; }
        @Override public void run() { if (list != null && !list.isEmpty()) logService.batchInsert(list); }
    }
}

Typical usage inside a transactional service method:

@Transactional(rollbackFor = Exception.class)
public boolean updateShippingDemandStatus(UpdateShippingDemandStatusReq req) {
    // todo something
    ServiceOrderLogUtils.saveLog(serviceOrderId, OperTypeEnum.CANCEL_SHIPPING_DEMAND, "用户取消运输需求");
    return true;
}

These snippets collectively demonstrate how to make message sending, distributed locking, and logging more reliable and less intrusive to the main business flow.