Java分布式环境下并发编程实践

本文涉及的产品
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云数据库 Redis 版,社区版 2GB
推荐场景:
搭建游戏排行榜
RDS MySQL Serverless 高可用系列,价值2615元额度,1个月
简介: Java分布式环境下并发编程实践

在Java中,我们可以使用线程来实现并发编程,但是在多线程编程中,我们需要考虑线程安全、锁、死锁等问题。本文将介绍Java中的并发编程,包括线程安全、锁、死锁等内容,同时提供实际的代码案例,让读者更容易理解和掌握。


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


569045c644f3aadcc792c479da213462.png



一、线程安全


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


解决线程安全问题的方法之一是使用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()方法。


三、死锁


image-20230409205211771.png

死锁是多线程编程中的一种问题,它发生在两个或多个线程互相等待对方释放锁的情况下。例如:

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锁,那么它们会等待一段时间,如果在这段时间内无法获取到对方的锁,就会自动释放自己的锁,避免死锁。


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


image-20230409205403525.png


在分布式系统中,由于不同的节点之间通过网络进行通信,因此会带来以下一些挑战:


1.网络延迟

在分布式系统中,由于节点之间通过网络进行通信,因此会存在网络延迟。这会导致节点之间的通信变慢,从而影响并发编程的效率。为了解决这个问题,可以采用异步编程模型,即通过回调函数的方式来处理网络通信。


2.数据一致性

在分布式系统中,由于数据分布在不同的节点上,因此会存在数据一致性的问题。如果不同节点上的数据不一致,就会导致系统出现异常。为了解决这个问题,可以采用分布式锁或者分布式事务来保证数据一致性。


3.容错性

在分布式系统中,由于节点之间存在网络通信,因此会存在节点宕机的情况。为了保证系统的容错性,需要采用一些容错机制,例如备份节点、自动故障转移等。


五、分布式锁的实现


在分布式系统中,为了保证数据一致性,需要采用分布式锁来控制对共享资源的访问。下面介绍一种基于Redis实现的分布式锁。

image-20230409205625391.png

笑小枫友情链接,感兴趣的进

更详细的分布式锁可以参考文章分布式锁的 3 种实现方案


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协议实现的分布式事务。


image-20230409205723358.png


笑小枫友情链接,感兴趣的进

具体事务相关可以参考文章七种常见分布式事务详解(2PC、3PC、TCC、Saga、本地事务表、MQ事务消息、最大努力通知)


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协议实现的分布式事务来保证数据一致性。

相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore &nbsp; &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&amp;RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
目录
相关文章
|
15小时前
|
Java 编译器 开发者
Java并发编程中的锁优化策略
【5月更文挑战第15天】 在Java的多线程编程中,锁机制是实现线程同步的关键。然而,不当的锁使用往往导致性能瓶颈甚至死锁。本文深入探讨了Java并发编程中针对锁的优化策略,包括锁粗化、锁消除、锁分离以及读写锁的应用。通过具体实例和性能分析,我们将展示如何有效避免竞争条件,减少锁开销,并提升应用程序的整体性能。
|
17小时前
|
Java 程序员 调度
Java中的多线程编程:从理论到实践
【5月更文挑战第14天】在现代计算机技术中,多线程编程是一个重要的概念。它允许多个线程并行执行,从而提高程序的运行效率。本文将从理论和实践两个角度深入探讨Java中的多线程编程,包括线程的基本概念、创建和控制线程的方法,以及如何处理线程同步和通信问题。
|
17小时前
|
存储 Java 分布式数据库
【分布式计算框架】HBase数据库编程实践
【分布式计算框架】HBase数据库编程实践
8 1
|
17小时前
|
分布式计算 并行计算 Java
【分布式计算框架】 MapReduce编程初级实践
【分布式计算框架】 MapReduce编程初级实践
7 2
|
17小时前
|
分布式计算 数据可视化 Hadoop
【分布式计算框架】HDFS常用操作及编程实践
【分布式计算框架】HDFS常用操作及编程实践
4 1
|
17小时前
|
数据采集 安全 Java
Java并发编程学习12-任务取消(上)
【5月更文挑战第6天】本篇介绍了取消策略、线程中断、中断策略 和 响应中断的内容
17 4
Java并发编程学习12-任务取消(上)
|
17小时前
|
Java 开发者
深入理解Java并发编程:从基础到高级
【5月更文挑战第13天】本文将深入探讨Java并发编程的各个方面,从基础知识到高级概念。我们将首先介绍线程的基本概念,然后深入讨论Java中的多线程编程,包括线程的创建和控制,以及线程间的通信。接下来,我们将探讨并发编程中的关键问题,如同步、死锁和资源竞争,并展示如何使用Java的内置工具来解决这些问题。最后,我们将讨论更高级的并发编程主题,如Fork/Join框架、并发集合和并行流。无论你是Java新手还是有经验的开发者,这篇文章都将帮助你更好地理解和掌握Java并发编程。
|
17小时前
|
Java
Java中的多线程编程:基础知识与实践
【5月更文挑战第13天】在计算机科学中,多线程是一种使得程序可以同时执行多个任务的技术。在Java语言中,多线程的实现主要依赖于java.lang.Thread类和java.lang.Runnable接口。本文将深入探讨Java中的多线程编程,包括其基本概念、实现方法以及一些常见的问题和解决方案。
|
17小时前
|
安全 算法 Java
深入理解Java并发编程:线程安全与性能优化
【5月更文挑战第13天】 在Java开发中,并发编程是一个复杂且重要的领域。它不仅关系到程序的线程安全性,也直接影响到系统的性能表现。本文将探讨Java并发编程的核心概念,包括线程同步机制、锁优化技术以及如何平衡线程安全和性能。通过分析具体案例,我们将提供实用的编程技巧和最佳实践,帮助开发者在确保线程安全的同时,提升应用性能。
10 1
|
17小时前
|
Java 编译器 开发者
Java并发编程中的锁优化策略
【5月更文挑战第13天】在Java并发编程中,锁是一种重要的同步机制,用于保证多线程环境下数据的一致性。然而,不当的使用锁可能会导致性能下降,甚至产生死锁等问题。本文将介绍Java中锁的优化策略,包括锁粗化、锁消除、锁降级等,帮助开发者提高程序的性能。

热门文章

最新文章