关于后端:Java分布式环境下并发编程实践

46次阅读

共计 10649 个字符，预计需要花费 27 分钟才能阅读完成。

在 Java 中，咱们能够应用线程来实现并发编程，然而在多线程编程中，咱们须要思考线程平安、锁、死锁等问题。本文将介绍 Java 中的并发编程，包含线程平安、锁、死锁等内容，同时提供理论的代码案例，让读者更容易了解和把握。

随着分布式系统越来越遍及，分布式系统中的并发编程成为了一个重要的话题。Java 作为一种高级编程语言，其并发编程能力失去了宽泛的认可。但在分布式系统中，Java 并发编程面临着一些新的挑战。本文将介绍在分布式系统下 Java 并发编程的一些技术和理论案例。

一、线程平安

在多线程编程中，线程平安是一个重要的问题。如果多个线程同时拜访同一个共享资源，就会呈现线程平安问题。例如，在银行账户转账时，如果多个线程同时对同一个账户进行操作，就会呈现线程平安问题。

解决线程平安问题的办法之一是应用 synchronized 关键字。synchronized 关键字能够将代码块或办法锁定，保障同一时间只有一个线程能够执行该代码块或办法。

上面是一个应用 synchronized 关键字的示例：

public class Counter {
    private int count;
    
    public synchronized void increment() {count++;}
    
    public synchronized void decrement() {count--;}
    
    public int getCount() {return count;}
}

在这个示例中，Counter 类有两个办法 increment()和 decrement()，它们都是应用 synchronized 关键字来保障线程平安。这样，同一时间只有一个线程能够执行 increment()和 decrement()办法。

二、锁

在 Java 中，锁是一种同步机制，能够用于管制多个线程对共享资源的拜访。Java 中的锁有两种类型：内置锁和显式锁。

内置锁是 Java 中的一个非凡对象，每个对象都有一个内置锁。能够应用 synchronized 关键字来获取内置锁。例如：

public synchronized void increment() {count++;}

在这个示例中，synchronized 关键字获取了 Counter 对象的内置锁。这样，在同一时间只有一个线程能够拜访 increment()办法。

显式锁是 Java 中的另一种锁类型，能够应用 java.util.concurrent.locks 包中的 Lock 接口来实现。与内置锁不同，显式锁提供了更多的灵活性和管制。例如：

public class Counter {
    private int count;
    private Lock lock = new ReentrantLock();
    
    public void increment() {lock.lock();
        try {count++;} finally {lock.unlock();
        }
    }
    
    public void decrement() {lock.lock();
        try {count--;} finally {lock.unlock();
        }
    }
    
    public int getCount() {return count;}
}

在这个示例中，Counter 类应用 ReentrantLock 类来创立一个显式锁。increment()和 decrement()办法获取锁并开释锁。这样，在同一时间只有一个线程能够拜访 increment()和 decrement()办法。

三、死锁

<img src=”https://image.xiaoxiaofeng.site/blog/image/image-20230409205211771.png?xiaoxiaofeng” alt=”image-20230409205211771″ style=”zoom:50%;” />

死锁是多线程编程中的一种问题，它产生在两个或多个线程相互期待对方开释锁的状况下。例如：

public class DeadlockExample {private final Object lock1 = new Object();
    private final Object lock2 = new Object();
    
    public void method1() {synchronized (lock1) {
            // do something
            synchronized (lock2) {// do something}
        }
    }
    
    public void method2() {synchronized (lock2) {
            // do something
            synchronized (lock1) {// do something}
        }
    }
}

在这个示例中，DeadlockExample 类有两个办法 method1()和 method2()，它们都应用两个锁 lock1 和 lock2。如果一个线程调用 method1()办法并获取了 lock1 锁，另一个线程调用 method2()办法并获取了 lock2 锁，那么两个线程都无奈继续执行，因为它们都在期待对方开释锁。这就是死锁。

防止死锁的办法之一是应用定时锁。定时锁能够在肯定工夫内主动开释锁，防止死锁。例如：

public class DeadlockExample {private final Object lock1 = new Object();
    private final Object lock2 = new Object();
    private final Lock timedLock1 = new ReentrantLock();
    private final Lock timedLock2 = new ReentrantLock();
    
    public void method1() {timedLock1.lock();
        try {
            // do something
            if (timedLock2.tryLock(500, TimeUnit.MILLISECONDS)) {
                try {// do something} finally {timedLock2.unlock();
                }
            }
        } catch (InterruptedException e) {Thread.currentThread().interrupt();} finally {timedLock1.unlock();
        }
    }
    
    public void method2() {timedLock2.lock();
        try {
            // do something
            if (timedLock1.tryLock(500, TimeUnit.MILLISECONDS)) {
                try {// do something} finally {timedLock1.unlock();
                }
            }
        } catch (InterruptedException e) {Thread.currentThread().interrupt();} finally {timedLock2.unlock();
        }
    }
}

在这个示例中，DeadlockExample 类应用定时锁 timedLock1 和 timedLock2 来防止死锁。如果一个线程调用 method1()办法并获取了 timedLock1 锁，另一个线程调用 method2()办法并获取了 timedLock2 锁，那么它们会期待一段时间，如果在这段时间内无奈获取到对方的锁，就会主动开释本人的锁，防止死锁。

四、分布式系统下的并发编程挑战

<img src=”https://image.xiaoxiaofeng.site/blog/image/image-20230409205403525.png?xiaoxiaofeng” alt=”image-20230409205403525″ style=”zoom:50%;” />

在分布式系统中，因为不同的节点之间通过网络进行通信，因而会带来以下一些挑战：

  1. 网络提早

在分布式系统中，因为节点之间通过网络进行通信，因而会存在网络提早。这会导致节点之间的通信变慢，从而影响并发编程的效率。为了解决这个问题，能够采纳异步编程模型，即通过回调函数的形式来解决网络通信。

  1. 数据一致性

在分布式系统中，因为数据分布在不同的节点上，因而会存在数据一致性的问题。如果不同节点上的数据不统一，就会导致系统出现异常。为了解决这个问题，能够采纳分布式锁或者分布式事务来保证数据一致性。

  1. 容错性

在分布式系统中，因为节点之间存在网络通信，因而会存在节点宕机的状况。为了保证系统的容错性，须要采纳一些容错机制，例如备份节点、主动故障转移等。

五、分布式锁的实现

在分布式系统中，为了保证数据一致性，须要采纳分布式锁来管制对共享资源的拜访。上面介绍一种基于 Redis 实现的分布式锁。

<img src=”https://image.xiaoxiaofeng.site/blog/image/image-20230409205625391.png?xiaoxiaofeng” alt=”image-20230409205625391″ style=”zoom:50%;” />

1. Redis 实现分布式锁的原理

Redis 是一个高性能的键值存储系统，反对多种数据结构，例如字符串、哈希表、列表等。Redis 提供了一种原子性的操作，能够实现分布式锁。

实现分布式锁的原理如下：

1）客户端向 Redis 发送一个 SETNX 命令，尝试去设置一个 key 的值为 1，如果这个 key 不存在，则设置胜利，否则设置失败。

2）客户端设置了这个 key 的值为 1 之后，就领有了这个锁。

3）其余客户端也能够向 Redis 发送 SETNX 命令，尝试去设置这个 key 的值为 1，然而因为这个 key 曾经存在了，因而设置失败。

4）当客户端实现了对共享资源的拜访之后，须要将这个 key 删除，以便其余客户端能够取得这个锁。

2. Redis 实现分布式锁的代码实现

上面是基于 Redis 实现分布式锁的代码实现：

public class RedisDistributedLock {
    private static final String LOCK_SUCCESS = "OK";
    private static final String SET_IF_NOT_EXIST = "NX";
    private static final String SET_WITH_EXPIRE_TIME = "PX";
    private static final Long RELEASE_SUCCESS = 1L;
    private static final String LOCK_PREFIX = "lock:";
    private JedisPool jedisPool;

    public RedisDistributedLock(JedisPool jedisPool) {this.jedisPool = jedisPool;}

    public boolean tryLock(String key, String requestId, int expireTime) {
        Jedis jedis = null;
        try {jedis = jedisPool.getResource();
            String lockKey = LOCK_PREFIX + key;
            String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
            if (LOCK_SUCCESS.equals(result)) {return true;}
        } finally {if (jedis != null) {jedis.close();
            }
        }
        return false;
    }

    public boolean releaseLock(String key, String requestId) {
        Jedis jedis = null;
        try {jedis = jedisPool.getResource();
            String lockKey = LOCK_PREFIX + key;
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
            if (RELEASE_SUCCESS.equals(result)) {return true;}
        } finally {if (jedis != null) {jedis.close();
            }
        }
        return false;
    }
}

3. Redisson 实现分布式锁的代码实现

当然，这里介绍的是比拟原生的形式，咱们也能够间接应用 Redisson 框架封装的分布式锁。

Redisson 是一个基于 Redis 的 Java 客户端，提供了丰盛的分布式数据结构和服务。其中就包含分布式锁的实现，上面介绍一下如何应用 Redisson 实现分布式锁。

  • 引入 Redisson 依赖
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.15.5</version>
</dependency>
  • 应用分布式锁

好的，上面提供一个更具体的代码示例：

import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

import java.util.concurrent.TimeUnit;

public class DistributedLockDemo {public static void main(String[] args) {
        // 创立 Redisson 客户端
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redisson = Redisson.create(config);

        // 获取分布式锁
        RLock lock = redisson.getLock("mylock");
        try {
            // 尝试获取锁，等待时间为 10 秒，锁的有效期为 5 秒
            boolean isLocked = lock.tryLock(10, 5, TimeUnit.SECONDS);
            if (isLocked) {
                // 获取锁胜利，执行业务逻辑
                System.out.println("获取锁胜利，执行业务逻辑...");
                Thread.sleep(3000); // 模仿业务逻辑耗时
            } else {
                // 获取锁失败，解决异常情况
                System.out.println("获取锁失败，解决异常情况...");
            }
        } catch (Exception e) {
            // 解决异常情况
            System.out.println("解决异常情况...");
        } finally {
            // 开释锁
            lock.unlock();
            System.out.println("开释锁...");
        }

        // 敞开 Redisson 客户端
        redisson.shutdown();}
}

在下面的代码中，咱们首先创立了一个 RedissonClient 对象，而后通过该对象获取一个 RLock 对象。在 try...catch...finally 代码块中，咱们调用 tryLock 办法尝试获取锁，如果获取胜利就执行业务逻辑；否则就解决异常情况。最初，在 finally 代码块中开释锁，并敞开 RedissonClient 对象。

须要留神的是，在理论利用中，咱们须要将下面的代码封装成一个可重入的分布式锁工具类，不便各个业务模块应用。

六、分布式事务的实现

在分布式系统中，为了保证数据一致性，须要采纳分布式事务来管制对共享资源的拜访。上面介绍一种基于 XA 协定实现的分布式事务。

<img src=”https://image.xiaoxiaofeng.site/blog/image/image-20230409205723358.png?xiaoxiaofeng” alt=”image-20230409205723358″ style=”zoom:33%;” />

1. XA 协定的原理

XA 协定是一种分布式事务协定，能够用于协调多个数据库的事务。XA 协定的原理如下：

1）事务管理器向数据库发送 XA START 命令，开始一个分布式事务。

2）事务管理器向数据库发送 XA END 命令，完结一个分布式事务。

3）事务管理器向数据库发送 XA PREPARE 命令，筹备提交一个分布式事务。

4）如果所有数据库都筹备好提交事务，则事务管理器向数据库发送 XA COMMIT 命令，提交分布式事务。

5）如果有任何一个数据库无奈提交事务，则事务管理器向所有数据库发送 XA ROLLBACK 命令，回滚分布式事务。

2. XA 协定的代码实现

上面是基于 XA 协定实现分布式事务的代码实现：

public class XADistributedTransaction {
    private static final String JDBC_DRIVER = "com.mysql.jdbc.Driver";
    private static final String DB_URL = "jdbc:mysql://localhost:3306/test";
    private static final String USER = "root";
    private static final String PASSWORD = "root";
    private static final String XA_DATASOURCE_CLASSNAME = "com.mysql.jdbc.jdbc2.optional.MysqlXADataSource";
    private static final String XA_DATASOURCE_URL = "jdbc:mysql://localhost:3306/test";
    private static final String XA_DATASOURCE_USER = "root";
    private static final String XA_DATASOURCE_PASSWORD = "root";
    private static final String XID_PREFIX = "xa_";
    private static final String TABLE_NAME = "account";
    private static final String TABLE_SCHEMA = "CREATE TABLE account (id INT PRIMARY KEY, balance INT)";
    private static final String INSERT_SQL = "INSERT INTO account (id, balance) VALUES (?, ?)";
    private static final String UPDATE_SQL = "UPDATE account SET balance = ? WHERE id = ?";

    public void transferMoney(int fromId, int toId, int amount) throws SQLException {XADataSource xaDataSource = getXADataSource();
        Connection connection = xaDataSource.getXAConnection().getConnection();
        Xid xid = createXid();
        try {connection.setAutoCommit(false);
            XAResource xaResource = getXAResource(connection);
            xaResource.start(xid, XAResource.TMNOFLAGS);
            try (PreparedStatement preparedStatement = connection.prepareStatement(UPDATE_SQL)) {preparedStatement.setInt(1, getBalance(connection, fromId) - amount);
                preparedStatement.setInt(2, fromId);
                preparedStatement.executeUpdate();}
            xaResource.end(xid, XAResource.TMSUCCESS);

            xaResource.start(xid, XAResource.TMNOFLAGS);
            try (PreparedStatement preparedStatement = connection.prepareStatement(UPDATE_SQL)) {preparedStatement.setInt(1, getBalance(connection, toId) + amount);
                preparedStatement.setInt(2, toId);
                preparedStatement.executeUpdate();}
            xaResource.end(xid, XAResource.TMSUCCESS);

            int prepare = xaResource.prepare(xid);
            if (prepare == XAResource.XA_OK) {xaResource.commit(xid, false);
            } else {xaResource.rollback(xid);
            }

            connection.commit();} catch (SQLException | XAException e) {connection.rollback();
            throw e;
        } finally {connection.close();
        }
    }

    private XADataSource getXADataSource() throws SQLException {MysqlXADataSource xaDataSource = new MysqlXADataSource();
        xaDataSource.setUrl(XA_DATASOURCE_URL);
        xaDataSource.setUser(XA_DATASOURCE_USER);
        xaDataSource.setPassword(XA_DATASOURCE_PASSWORD);
        return xaDataSource;
    }

    private Xid createXid() throws XAException {byte[] gtrid = new byte[10];
        byte[] bqual = new byte[10];
        Arrays.fill(gtrid, (byte) 9);
        Arrays.fill(bqual, (byte) 9);
        return new XidImpl(0x1234, gtrid, bqual);
    }

    private XAResource getXAResource(Connection connection) throws SQLException {return connection.unwrap(XAResource.class);
    }

    private int getBalance(Connection connection, int id) throws SQLException {try (PreparedStatement preparedStatement = connection.prepareStatement("SELECT balance FROM account WHERE id = ?")) {preparedStatement.setInt(1, id);
            try (ResultSet resultSet = preparedStatement.executeQuery()) {if (resultSet.next()) {return resultSet.getInt("balance");
                }
            }
        }
        throw new RuntimeException("Account not found:" + id);
    }

    public void init() throws SQLException {try (Connection connection = DriverManager.getConnection(DB_URL, USER, PASSWORD)) {try (Statement statement = connection.createStatement()) {statement.executeUpdate("DROP TABLE IF EXISTS account");
                statement.executeUpdate(TABLE_SCHEMA);
                try (PreparedStatement preparedStatement = connection.prepareStatement(INSERT_SQL)) {preparedStatement.setInt(1, 1);
                    preparedStatement.setInt(2, 1000);
                    preparedStatement.executeUpdate();}
                try (PreparedStatement preparedStatement = connection.prepareStatement(INSERT_SQL)) {preparedStatement.setInt(1, 2);
                    preparedStatement.setInt(2, 1000);
                    preparedStatement.executeUpdate();}
            }
        }
    }
}

七、总结

本文介绍了 Java 中的并发编程，包含线程平安、锁、死锁等内容。在多线程编程中，线程平安是一个重要的问题，能够应用 synchronized 关键字或显式锁来实现。死锁是一个常见的问题，能够应用定时锁来防止。多线程编程须要认真思考线程平安和锁的问题，能力保障程序的正确性和性能。

并且解说了在分布式系统下 Java 并发编程的一些技术和理论案例。在分布式系统中，Java 并发编程须要面对网络提早、数据一致性和容错性等挑战，须要采纳一些技术和机制来解决这些问题。例如，能够采纳基于 Redis 实现的分布式锁来管制对共享资源的拜访，也能够采纳基于 XA 协定实现的分布式事务来保证数据一致性。

正文完
后端
发表至： 后端
2023-04-09
 0
关于后端:给Hexo博客文章加密
关于后端:如何理解JAVA中的泛型
关于后端:COMP24011研究
关于后端:正月初五迎财神