那些年，我们见过的Java服务端“问题”

导读

道无精粗，人之所见有精粗。如这一间房，人初进来，只见一个大规模如此。处久，便柱壁之类，一一看得明白。再久，如柱上有些文藻，细细都看出来。然只是一间房。

1.系统不是分布式

1.1.单机版系统抢单案例

// 抢取订单函数
public synchronized void grabOrder(Long orderId, Long userId) {
    // 获取订单信息
    OrderDO order = orderDAO.get(orderId);
    if (Objects.isNull(order)) {
        throw new BizRuntimeException(String.format("订单(%s)不存在", orderId));
    }
    // 检查订单状态
    if (!Objects.equals(order.getStatus, OrderStatus.WAITING_TO_GRAB.getValue())) {
        throw new BizRuntimeException(String.format("订单(%s)已被抢", orderId));
    }
    
    // 设置订单被抢
    orderDAO.setGrabed(orderId, userId);
}

1.2.分布式系统抢单案例

// 抢取订单函数
public void grabOrder(Long orderId, Long userId) {
    Long lockId = orderDistributedLock.lock(orderId);
    try {
        grabOrderWithoutLock(orderId, userId);
    } finally {
        orderDistributedLock.unlock(orderId, lockId);
    }
}
// 不带锁的抢取订单函数
private void grabOrderWithoutLock(Long orderId, Long userId) {
    // 获取订单信息
    OrderDO order = orderDAO.get(orderId);
    if (Objects.isNull(order)) {
        throw new BizRuntimeException(String.format("订单(%s)不存在", orderId));
    }
    // 检查订单状态
    if (!Objects.equals(order.getStatus, OrderStatus.WAITING_TO_GRAB.getValue())) {
        throw new BizRuntimeException(String.format("订单(%s)已被抢", orderId));
    }
    
    // 设置订单被抢
    orderDAO.setGrabed(orderId, userId);
}

1.3.分布式系统的优缺点

1. 可靠性、高容错性：

1. 可扩展性：

1. 灵活性：

1. 性能高：

1. 性价比高：

1. 排查难度高：

1. 软件支持少：

1. 建设成本高：

1.4.分布式软件解决方案

1.4.1.分布式锁

1. 基于数据库实现的分布式锁；
2. 基于redis实现的分布式锁；
3. 基于Zookeeper实现的分布式锁。

1.4.2.分布式消息

1.4.3.数据库分片分组

1.4.4.分布式计算

1.5.分布式硬件部署方案

1.5.1.一般单机版部署方案

1.5.2.中小型分布式硬件部署方案

1.5.3.大型分布式硬件部署方案

2.多线程使用不正确

2.1.一个慢接口案例

// 登录函数(示意写法)
public UserVO login(String phoneNumber, String verifyCode) {
    // 检查验证码
    if (!checkVerifyCode(phoneNumber, verifyCode)) {
        throw new ExampleException("验证码错误");
    }
    // 检查用户存在
    UserDO user = userDAO.getByPhoneNumber(phoneNumber);
    if (Objects.nonNull(user)) {
        return transUser(user);
    }
    // 创建新用户
    return createNewUser(user);
}
// 创建新用户函数
private UserVO createNewUser(String phoneNumber) {
    // 创建新用户
    UserDO user = new UserDO();
    ...
    userDAO.insert(user);
    // 绑定优惠券
    couponService.bindCoupon(user.getId(), CouponType.NEW_USER);
    
    // 返回新用户
    return transUser(user);
}

2.2.采用多线程优化

// 创建新用户函数
private UserVO createNewUser(String phoneNumber) {
    // 创建新用户
    UserDO user = new UserDO();
    ...
    userDAO.insert(user);
    // 绑定优惠券
    executorService.execute(()->couponService.bindCoupon(user.getId(), CouponType.NEW_USER));
    
    // 返回新用户
    return transUser(user);
}

2.3.采用消息队列优化

// 创建新用户函数
private UserVO createNewUser(String phoneNumber) {
    // 创建新用户
    UserDO user = new UserDO();
    ...
    userDAO.insert(user);
    // 发送优惠券消息
    Long userId = user.getId();
    CouponMessageDataVO data = new CouponMessageDataVO();
    data.setUserId(userId);
    data.setCouponType(CouponType.NEW_USER);
    Message message = new Message(TOPIC, TAG, userId, JSON.toJSONBytes(data));
    SendResult result = metaqTemplate.sendMessage(message);
    if (!Objects.equals(result, SendStatus.SEND_OK)) {
        log.error("发送用户({})绑定优惠券消息失败:{}", userId, JSON.toJSONString(result));
    }
    // 返回新用户
    return transUser(user);
}

// 优惠券服务类
@Slf4j
@Service
public class CouponService extends DefaultMessageListener {
    // 消息处理函数
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void onReceiveMessages(MetaqMessage message) {
        // 获取消息体
        String body = message.getBody();
        if (StringUtils.isBlank(body)) {
            log.warn("获取消息({})体为空", message.getId());
            return;
        }
        
        // 解析消息数据
        CouponMessageDataVO data = JSON.parseObject(body, CouponMessageDataVO.class);
        if (Objects.isNull(data)) {
            log.warn("解析消息({})体为空", message.getId());
            return;
        }
        // 绑定优惠券
        bindCoupon(data.getUserId(), data.getCouponType());
    }
}

1. 如果系统发生重启或崩溃，导致消息处理函数执行失败，不会确认消息已消费；由于MetaQ支持多服务订阅同一队列，该消息可以转到别的服务进行消费，亦或等到本服务恢复正常后再进行消费。
2. 消费者可多服务、多线程进行消费消息，即便消息处理时间较长，也不容易引起消息积压；即便引起消息积压，也可以通过扩充服务实例的方式解决。
3. 如果需要重新消费该消息，只需要在MetaQ管理平台上点击"消息验证"即可。

3.流程定义不合理

3.1.原有的采购流程

/** 完成采购动作函数(此处省去获取采购单/验证状态/锁定采购单等逻辑) */
public void finishPurchase(PurchaseOrder order) {
    // 完成相关处理
    ......
    // 回流采购单(调用HTTP接口)
    backflowPurchaseOrder(order);
    
    // 设置完成状态
    purchaseOrderDAO.setStatus(order.getId(), PurchaseOrderStatus.FINISHED.getValue());
}

1. 该函数可能耗费时间较长，导致完成采购接口成为慢接口；
2. 该函数可能失败抛出异常，导致客户调用完成采购接口失败。

3.2.优化的采购流程

/** 完成采购动作函数(此处省去获取采购单/验证状态/锁定采购单等逻辑) */
public void finishPurchase(PurchaseOrder order) {
    // 完成相关处理
    ......
    
    // 设置完成状态
    purchaseOrderDAO.setStatus(order.getId(), PurchaseOrderStatus.FINISHED.getValue());
}
/** 执行回流动作函数(此处省去获取采购单/验证状态/锁定采购单等逻辑) */
public void executeBackflow(PurchaseOrder order) {
    // 回流采购单(调用HTTP接口)
    backflowPurchaseOrder(order);
    
    // 设置回流状态
    purchaseOrderDAO.setStatus(order.getId(), PurchaseOrderStatus.BACKFLOWED.getValue());
}

3.3.有限状态机介绍

3.3.1.概念

3.3.2.要素

3.3.3.状态

1. 状态必须是一个持久状态，而不能是一个临时状态；
2. 终结状态不能是中间状态，不能继续进行流程流转；
3. 状态划分合理，不要把多个状态强制合并为一个状态；
4. 状态尽量精简，同一状态的不同情况可以用其它字段表示。

3.3.4.动作

1. 每个动作执行前，必须检查当前状态和触发动作状态的一致性；
2. 状态机的状态更改，只能通过动作进行，其它操作都是不符合规范的；
3. 需要添加分布式锁保证动作的原子性，添加数据库事务保证数据的一致性；
4. 类似的动作（比如操作用户、请求参数、动作含义等）可以合并为一个动作，并根据动作执行结果转向不同的状态。

4.系统间交互不科学

4.1.直接通过数据库交互

/** 执行回流动作函数(此处省去获取采购单/验证状态/锁定采购单等逻辑) */
public void executeBackflow(PurchaseOrder order) {
    // 完成原始采购单
    rawPurchaseOrderDAO.setStatus(order.getRawId(), RawPurchaseOrderStatus.FINISHED.getValue());
    
    // 设置回流状态
    purchaseOrderDAO.setStatus(order.getId(), PurchaseOrderStatus.BACKFLOWED.getValue());
}

1. 直接暴露数据库表，容易产生数据安全问题；
2. 多个系统操作同一数据库表，容易造成数据库表数据混乱；
3. 操作同一个数据库表的代码，分布在不同的系统中，不便于管理和维护；
4. 具有数据库表这样的强关联，无法实现系统间的隔离和解耦。

4.2.通过Dubbo接口

/** 采购单服务接口 */
public interface PurchaseOrderService {
    /** 完成采购单函数 */
    public void finishPurchaseOrder(Long orderId);
}
/** 采购单服务实现 */
@Service("purchaseOrderService")
public class PurchaseOrderServiceImpl implements PurchaseOrderService {
    /** 完成采购单函数 */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void finishPurchaseOrder(Long orderId) {
        // 相关处理
        ...
        // 完成采购单
        purchaseOrderService.finishPurchaseOrder(order.getRawId());
    }
}

/** 执行回流动作函数(此处省去获取采购单/验证状态/锁定采购单等逻辑) */
public void executeBackflow(PurchaseOrder order) {
    // 完成采购单
    purchaseOrderService.finishPurchaseOrder(order.getRawId());
    
    // 设置回流状态
    purchaseOrderDAO.setStatus(order.getId(), PurchaseOrderStatus.BACKFLOWED.getValue());
}

4.3.常见系统间交互协议

4.3.1.同步接口调用

1. HTTP/HTTPS接口；
2. WebService接口；
3. Dubbo/HSF接口；
4. CORBA接口。

4.3.2.异步消息通知

1. MetaQ的消息通知；
2. CORBA消息通知。

4.4.常见系统间交互方式

4.4.1.请求-应答

4.4.2.通知-确认

4.4.3.请求-应答-查询-返回

4.4.4.请求-应答-回调

4.4.5.请求-应答-通知-确认

4.4.6.通知-确认-通知-确认

5.数据查询不分页

5.1.普通查询案例

/** 订单DAO接口 */
public interface OrderDAO {
    /** 查询过期订单函数 */
    @Select("select * from t_order where status = 5 and gmt_create < date_sub(current_timestamp, interval 30 day)")
    public List queryTimeout();
}
/** 订单服务接口 */
public interface OrderService {
    /** 查询过期订单函数 */
    public List queryTimeout();
}

1. 数据量太大，导致服务端的内存溢出；
2. 数据量太大，导致查询接口超时、返回数据超时等；
3. 数据量太大，导致客户端的内存溢出。

5.2.设置最大数量

/** 订单DAO接口 */
public interface OrderDAO {
    /** 查询过期订单函数 */
    @Select("select * from t_order where status = 5 and gmt_create < date_sub(current_timestamp, interval 30 day) limit 0, #{maxCount}")
    public List queryTimeout(@Param("maxCount") Integer maxCount);
}
/** 订单服务接口 */
public interface OrderService {
    /** 查询过期订单函数 */
    public List queryTimeout(Integer maxCount);
}

5.3.采用分页查询

/** 订单DAO接口 */
public interface OrderDAO {
    /** 统计过期订单函数 */
    @Select("select count(*) from t_order where status = 5 and gmt_create < date_sub(current_timestamp, interval 30 day)")
    public Long countTimeout();
    /** 查询过期订单函数 */
    @Select("select * from t_order where status = 5 and gmt_create < date_sub(current_timestamp, interval 30 day) limit #{startIndex}, #{pageSize}")
    public List queryTimeout(@Param("startIndex") Long startIndex, @Param("pageSize") Integer pageSize);
}
/** 订单服务接口 */
public interface OrderService {
    /** 查询过期订单函数 */
    public PageData queryTimeout(Long startIndex, Integer pageSize);
}

5.3.分页查询隐藏问题

/** 订单DAO接口 */
public interface OrderDAO {
    /** 查询过期订单函数 */
    @Select("select * from t_order where status = 5 and gmt_create < date_sub(current_timestamp, interval 30 day) limit #{startIndex}, #{pageSize}")
    public List queryTimeout(@Param("startIndex") Long startIndex, @Param("pageSize") Integer pageSize);
    /** 设置订单超时关闭 */
    @Update("update t_order set status = 10 where id = #{orderId} and status = 5")
    public Long setTimeoutClosed(@Param("orderId") Long orderId)
}
/** 关闭过期订单作业类 */
public class CloseTimeoutOrderJob extends Job {
    /** 分页数量 */
    private static final int PAGE_COUNT = 100;
    /** 分页大小 */
    private static final int PAGE_SIZE = 1000;
    /** 作业执行函数 */
    @Override
    public void execute() {
        for (int i = 0; i < PAGE_COUNT; i++) {
            // 查询处理订单
            List orderList = orderDAO.queryTimeout(i * PAGE_COUNT, PAGE_SIZE);
            for (OrderDO order : orderList) {
                // 进行超时关闭
                ......
                orderDAO.setTimeoutClosed(order.getId());
            }
            // 检查处理完毕
            if(orderList.size() < PAGE_SIZE) {
                break;
            }
        }
    }
}

当满足查询条件的数据，在操作中不再满足查询条件时，会导致后续分页查询中前startIndex（开始序号）条满足条件的数据被跳过。

/** 订单DAO接口 */
public interface OrderDAO {
    /** 查询过期订单函数 */
    @Select("select * from t_order where status = 5 and gmt_create < date_sub(current_timestamp, interval 30 day) limit 0, #{maxCount}")
    public List queryTimeout(@Param("maxCount") Integer maxCount);
    /** 设置订单超时关闭 */
    @Update("update t_order set status = 10 where id = #{orderId} and status = 5")
    public Long setTimeoutClosed(@Param("orderId") Long orderId)
}
/** 关闭过期订单作业(定时作业) */
public class CloseTimeoutOrderJob extends Job {
    /** 分页数量 */
    private static final int PAGE_COUNT = 100;
    /** 分页大小 */
    private static final int PAGE_SIZE = 1000;
    /** 作业执行函数 */
    @Override
    public void execute() {
        for (int i = 0; i < PAGE_COUNT; i++) {
            // 查询处理订单
            List orderList = orderDAO.queryTimeout(PAGE_SIZE);
            for (OrderDO order : orderList) {
                // 进行超时关闭
                ......
                orderDAO.setTimeoutClosed(order.getId());
            }
            // 检查处理完毕
            if(orderList.size() < PAGE_SIZE) {
                break;
            }
        }
    }
}

后记