分布式交易系统的并发处理, 以及用Redis和Zookeeper实现分布式锁-白红宇

分布式交易系统的并发处理, 以及用Redis和Zookeeper实现分布式锁

阅读量：6924 次

发布时间：2019-06-27

本文共 9457 字，大约阅读时间需要 31 分钟。

交易系统

交易系统的数据结构

支付系统API通常需要一个“订单号”作为入参, 而实际调用API接口时使用到的往往不是真正意义的业务订单号, 而是交易订单号. 支付系统的API会使用“商户号+订单号”唯一的标准来设计, 对于商户方就需要做对应的逻辑来保证业务的一致性. 这里就引入了交易订单表, 一个业务订单在支付时会创建一条交易订单，这笔交易订单会关联业务订单，并将交易订单号发给支付系统, 根据结果处理资金账户数据和业务订单数据. 由于是调用远程接口, 有同步也有异步, 结果会出现各种各样的情况，如成功, 失败, 等待, 超时等等, 因此一笔业务订单可能对应多条交易订单，每一条交易订单会对应一个或多个请求结果.

交易的模式

交易主要有三个模式: 同步请求, 异步请求, 还有查询. 对于一些系统, 还有批量提交的请求方式, 这个可以归为异步请求这一类.

对于同步的交易, 可能会在发出请求后收到成功, 失败, 未知三种情况;

对于异步的交易, 可能会收到成功, 失败两种情况;

对于查询, 可能会收到成功, 失败, 未知三种情况, 和同步交易一样.

交易的结果处理

成功: 更新交易订单状态, 记录结果, 根据实际业务处理.

失败: 更新交易订单状态, 记录结果, 根据实际业务, 创建新的交易订单或者将业务订单置为失败.

未知: 不做操作, 等待异步通知, 或通过时间任务异步查询, 或加入队列进行异步查询.

需要注意的是, 对于有多种返回结果代码的支付系统, 一定要明确各个代码的归类, 属于"成功"和"失败"的代码不能出现偏差. 在通道方的结果代码有调整时, 要及时更新.

交易的事务控制

交易的事务应当仅仅局限于本地方法, 中间不能有远程调用, 因为远程接口不可控, 更不可能在事务失败时跟随本地调用一起回滚. 另外还可能导致本地资源一直被占用, 尤其是数据库连接.

并发问题

单机的并发可以通过synchronized或者Lock解决(全局一致性), 也可以通过乐观锁解决(最终一致性), 同时使用队列降低系统突发压力. 这个比较简单就不说明了.

对于分布式系统的并发, 可以通过以下途径解决:

乐观锁

乐观锁是通过数据库入库时, 校验数据版本的一致性来达到业务最终一致性的一种手段, 适用于单机分布式等各种环境, 好处是实现简单, 读性能非常好, 缺点也很明显, 在业务的交易链较长时, 一个回滚可能会导致整个上层交易失败, 这样的情况虽然能保证资金不出错, 但是重试的概率增大后, 也是会影响业务效率的.

分布式锁

分布式锁可以基于db, redis, zookeeper等实现. 最简单的锁实现的是lock和unlock功能, 实际应用中, 还需要两个功能: 一个Reentrant 以实现同线程重入, 和一个Timeout 以实现在某个实例出现异常时, 不至于导致整个交易被永久挂起.

性能优化

性能优化是通过队列缓冲突发负载, 和排重减少实际交易的请求来进行的.

Redisson的分布式锁

借助Redisson的getLock和getReadWriteLock方法, 对同线程可重入, 可以设置锁超时, 可以设置取锁超时, 并且锁本身有默认30秒的超时

public class LockManagerImpl implements LockManager {    private final Redisson redisson;    public LockManagerImpl(ZookeeperManager zookeeperManager) {        Map
     
       settings = zookeeperManager.load("/lock");        Config config = new Config();        config                .useSingleServer()                .setAddress(settings.get("address").getString("redis://127.0.0.1:6379"))                .setTimeout(settings.get("timeout").getInteger(3000))                .setPassword(settings.get("password").getString(null));        redisson = (Redisson) Redisson.create(config);    }    public void init() {        logger.debug("init()");    }    public void destroy() {        logger.debug("destroy()");    }    @Override    public Lock getLock(String key) {        return redisson.getLock(key);    }    @Override    public ReadWriteLock getReadWriteLock(String key) {        return redisson.getReadWriteLock(key);    }}

Jedis实现的分布式锁

借助 SETNX 命令, 只有当key不存在时才能set成功, 这只是一个简单的实现, 有超时, 但是不能同一线程重入.

@Component("redisLockHelper")public class RedisLockHelper {    final Logger logger = LoggerFactory.getLogger(this.getClass());    @Autowired    private StringRedisTemplate jedis;    /**     * 获取一个redis锁（根据key）     * (jedis.opsForValue()没有setNX接口则通过execute实现)     * @param lockName 锁名称     * @param value 锁其它信息     * @param expire 锁过期时间 (单位：秒)     * @return 是否成功获取锁：true-成功获取锁，false-获取锁失败      */    public boolean lock(final String lockName,final String value, final long expire) {        try {            final byte[] lockBytes = jedis.getStringSerializer().serialize(lockName);            final byte[] valueBytes = jedis.getStringSerializer().serialize(value);            Long rs= jedis.execute(new RedisCallback
     
      () {                public Long doInRedis(RedisConnection connection) {                    boolean locked = connection.setNX(lockBytes, valueBytes);                    if (locked) {                        connection.expire(lockBytes, expire);//设置一个过期时间                        return 1L;                    }                    //未获取到锁则ttl检查过期时间（防止setNX、expire间的崩溃）                    Long checkExpire=connection.ttl(lockBytes);                    if(checkExpire == -1){//ttl：-1, 如果key没有到期超时。 -2, 如果键不存在。                         connection.expire(lockBytes, expire);//如果以前的key没有过期机制，则设置一个过期时间                        logger.warn("ttl检查有效，重新设置失效时间! [lockName={}, expire={}]", lockName, expire);                    }                    return 0L;//不做排队处理，直接返回失败                }            });            return (rs==1L);        } catch (Exception e) {            logger.error("获取redis锁异常! [lockName="+lockName+"]",e);        }        return false;    }    public void unlock(final String lockName) {        try {            jedis.delete(lockName);        } catch (Exception e) {            logger.error("释放redis锁异常! [lockName="+lockName+"]",e);        }    }}

Zookeeper实现的分布式锁

利用了Zookeeper的Watcher机制. 在Zookeeper中节点类型使用 EPHEMERAL_SEQUENTIAL, 这种类型当客户端无效后会自动删除, 并且同名节点会通过后缀数字增长进行添加. 这样实际上维护了两个序列: 在Zookeeper中会保持一个同名但是后缀数字不断增长的序列, 而在本地是线程序列, 使用一个同步的lock对getChildren进行竞争. 每一个本地线程都会在zookeeper中创建一个带序列号的节点, 同时等待资源锁被释放, 当拿到资源锁时, 判断自己是不是top的那个节点, 如果不是就释放资源锁, 继续等待. 如果是就说明拿到业务锁了, 在业务执行完之后, 要调用unlock释放业务锁, 触发watcher事件. 如果拿到业务锁的线程中途退出了并未执行unlock, zookeeper在检查到客户节点退出后, 也会将对应的节点删除, 也会触发watcher事件.

public class DistributedLock {    private final ZooKeeper zk;    private final String lockBasePath;    private final String lockName;    private String lockPath;    public DistributedLock(ZooKeeper zk, String lockBasePath, String lockName) {        this.zk = zk;        this.lockBasePath = lockBasePath;        this.lockName = lockName;    }    public void lock() throws IOException {        try {            // lockPath will be different than (lockBasePath + "/" + lockName) becuase of the sequence number ZooKeeper appends            lockPath = zk.create(lockBasePath + "/" + lockName, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);            final Object lock = new Object();            // The requests in the same jvm will be blocked here waiting for wait() or notifyAll(). This will prevent missing notifications.            synchronized(lock) {                while(true) {                    List
     
       nodes = zk.getChildren(lockBasePath, new Watcher() {                        @Override                        public void process(WatchedEvent event) {                            synchronized (lock) {                                // When the brother nodes are changed, all waiting threads will be notified.                                lock.notifyAll();                            }                        }                    });                    Collections.sort(nodes); // ZooKeeper node names can be sorted lexographically                    if (lockPath.endsWith(nodes.get(0))) {                        return;                    } else {                        // This will give up the lock and wait the next notification. When woken up, it will go through the WHILE block again                        lock.wait();                    }                }            }        } catch (KeeperException e) {            throw new IOException (e);        } catch (InterruptedException e) {            throw new IOException (e);        }    }    public void unlock() throws IOException {        try {            // This will trigger the Watcher.process()            zk.delete(lockPath, -1);            lockPath = null;        } catch (KeeperException e) {            throw new IOException (e);        } catch (InterruptedException e) {            throw new IOException (e);        }    }}

Jedis实现的队列

利用 Redis 的 LIST 类型数据的 RPUSH 和 BLPOP 方法实现消息的生产和消费

public long rpush(final String... value) {    if (value == null) return -1;    return (Long) execute((Jedis jedis) -> jedis.rpush(getId(), value));}public long rpushObject(final Object value) {    if (value == null) return -1;    return (Long) execute((Jedis jedis) -> jedis.rpush(getId().getBytes(), SerializeUtil.serialize(value)));}public long rpushObject(final Object... value) {    if (value == null || value.length == 0) return -1;    return (Long) execute((Jedis jedis) -> jedis.rpush(getId().getBytes(), SerializeUtil.serialize(value)));}public List
     
       blpop(int timeout) {    return (List
      
       ) execute((Jedis jedis)-> jedis.blpop(timeout, getId()));}public List

业务中使用队列

@Overridepublic long lRpush(String id, String value) {    return factory.getList(id).rpush(value);}@Overridepublic long lRpushObject(String id, Object value) {    return factory.getList(id).rpushObject(value);}@Overridepublic List
     
       lBlpop(String id, int timeout) {    return factory.getList(id).blpop(timeout);}@Overridepublic List

转载地址：http://hhhcl.baihongyu.com/

你可能感兴趣的文章