Java并发编程-撸一个数据库连接池

简介: 章节目录等待超时模式的使用场景可以掌握的技能等待/通知 消费者/生产者模式CountDownLatch、AtomicInteger、静态内部类、LinkedList、动态代理的使用1.等待超时模式场景当我们调用方法时,这个方法返回的资源比较重要,比如获取数据库连接池中连接句柄。

章节目录

  • 等待超时模式的使用场景
  • 可以掌握的技能
    • 等待/通知 消费者/生产者模式
    • CountDownLatch、AtomicInteger、静态内部类、LinkedList、动态代理的使用

1.等待超时模式

场景

当我们调用方法时,这个方法返回的资源比较重要,比如获取数据库连接池中连接句柄。但是这个资源的返回随着业务量的增加,那么获取资源(连接池连接)的时间就会增加,那么调用一个方法时就要等待一段时间(一般来说是给定一个时间段),如果该方法能够在一段时间内获取到结果,那么将结果立刻返回,反之,超时返回默认结果。

等待/通知的经典范式,即加锁、条件循环、处理逻辑3个步骤,这种范式没办法做到超时等待,对经典范式做很小的改动,就可以实现超时等待。

等待超时模式伪代码:

   public synchronized Object get(long mills) throws InterruptedException {
       Object result = null;
       long future = System.currentTimeMills() + mills;
       long remaining = mills;
       while (result == null && remaining > 0) {
            wait(remaining);//释放锁,阻塞 mills 毫秒
            remaining = future - System.currentTimeMills();
       }

       return result;//如果超时之后获取到result则不返回null
   }

超时等待的作用就是不会永远阻塞调用者,但是 超时之后被唤醒,知识将线程从等待队列移动至阻塞队列,继续向下进行返回result还是要重新获取锁,如果一直获取不到锁,那么result也不会打印。只是增加了灵活性。

2.可以掌握的技能

实战
使用等待超时模式撸一个简单数据库连接池,在示例中模拟:

  • 从连接池获取连接 RunnerThread
  • 使用连接 RunnerThread
  • 释放连接 RunnerThread
    注意:客户端获取(消费)连接的过程被设定为等待超时、等待/通知两种模式
    ConnectionPool.java-数据库连接池
package org.seckill.DBConnection;

import java.sql.Connection;
import java.util.LinkedList;

/**
 * 数据库连接池对象
 */
public class ConnectionPool {

    //链表list(池)维护 connection 连接对象
    private LinkedList<Connection> pool = new LinkedList<Connection>();

    //构造方法 初始化池中连接
    public ConnectionPool(int initialSize) {
        if (initialSize > 0) {
            for (int i = 0; i < initialSize; i++) {
                pool.addLast(ConnectionDriver.createConnection());//创建initialSize个代理Connection对象
            }
        }
    }

    //释放connection ,相当于-生产者
    public void releaseConnection(Connection connection) {
        if (connection != null) {//有效归还连接
            synchronized (pool) {
                pool.addLast(connection);
                //生产者动作完毕后,需要唤醒所有消费者
                pool.notifyAll();
            }
        }
    }

    //获取connection句柄,相当于消费者,采用超时等待与等待/通知两种策略
    public Connection fetchConnection(long mills) throws InterruptedException {
        synchronized (pool) {
            if (mills < 0) {//非超时等待模式,采用等待/通知模式
                while (pool.isEmpty()) {
                    pool.wait();//本示例中不演示这种模式下获取连接的情景
                }

                return pool.removeFirst();
            } else {//超时等待模式
                long future = System.currentTimeMillis() + mills;
                long remaining = mills;
                while (pool.isEmpty() && remaining > 0) {
                    pool.wait(remaining);
                    remaining = future - System.currentTimeMillis();
                }

                Connection connection = null;

                if (!pool.isEmpty()) {
                    connection = pool.removeFirst();//返回头结点对象
                }

                return connection;
            }
        }
    }

}

ConnectionDriver.java-动态生成Connection代理对象

package org.seckill.DBConnection;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.sql.Connection;

/**
 * 数据库连接驱动,
 * 动态代理获取实现java.sql.Connection 接口的代理对象
 */
public class ConnectionDriver {

    static class ConnectionHandler implements InvocationHandler {
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
            if (method.getName() == "commit") {
                Thread.sleep(100);
            }
            return null;
        }

    }

    //获取Connection的动态代理类
    public static final Connection createConnection() {
        return (Connection) Proxy.newProxyInstance(
                ConnectionDriver.class.getClassLoader(),//类加载器
                new Class<?>[]{Connection.class},//Connection实现的接口列表,包含Connection接口
                new ConnectionHandler());//与代理对象绑定的handler
    }
}

ConnectionPoolTest.java--测试类

package org.seckill.DBConnection;

import java.sql.Connection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

public class ConnectionPoolTest {
    //线程池中初始化10个连接
    static ConnectionPool connectionPool = new ConnectionPool(10);
    //保证所有的ConnectionRunner 能够同时开始
    static CountDownLatch start = new CountDownLatch(1);
    //main线程将等待所有的Connection Runner结束后才开始执行
    static CountDownLatch end;

    public static void main(String[] args) throws Exception {
        //ConnectionRunner 线程数量,可以修改线程数量进行观察
        int threadCount = 50;
        end = new CountDownLatch(threadCount);
        int count = 20;//每个线程进行20次fetchConnetion动作
        AtomicInteger got = new AtomicInteger();
        AtomicInteger notGot = new AtomicInteger();

        for (int i = 0; i < threadCount; i++) {
            Thread thread = new Thread(new ConnectionRunner(count, got, notGot), "ConnectionRunnerThread");
            thread.start();
        }

        start.countDown();//使所有线程同时运行
        end.await();//主线程等待所有线程运行完
        System.out.println("总的请求次数" + threadCount * count);
        System.out.println("获取到的连接总数" + got);
        System.out.println("未获取到的连接总数" + notGot);


    }

    static class ConnectionRunner implements Runnable {
        int count;//每个线程fetchConnetion的次数
        AtomicInteger got;//记录fetchConnection 成功的次数
        AtomicInteger notGot;//记录fetchConnetion 未成功的次数

        public ConnectionRunner(int count, AtomicInteger got, AtomicInteger notGot) {
            this.count = count;
            this.got = got;
            this.notGot = notGot;
        }

        public void run() {
            try {
                start.await();//等待 所有ConnectionRunner 初始化成功且处于Runnable状态,同时开始运行,由主线程控制的
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            while (count > 0) {
                try {
                    //从连接池中获取连接,如果1000ms内无法获取到,将会返回null。
                    Connection connection = connectionPool.fetchConnection(1000);
                    if (connection != null) {
                        try {
                            connection.createStatement();
                            connection.commit();
                        } finally {
                            //归还连接
                            connectionPool.releaseConnection(connection);
                            got.incrementAndGet();//对获取次数状态进行更改
                        }
                    } else {
                        notGot.incrementAndGet();//对未获取次数状态进行更改
                    }
                } catch (Exception e) {

                } finally {
                    count--;//运行次数递减
                }
            }

            end.countDown();
        }
    }
}

运行结果

1.设置RunnerConnection threadCount数为10

img_610e3d11cb7f5a62475b64a11a3c4cad.png
threadCount = 10

2.设置RunnerConnection threadCount数为20

img_8018c1843e67af0217bfb1c42be5e333.png
threadCount = 20

3.设置RunnerConnection threadCount数为50

img_6c3db5e7732a40ddd18fc8cdd5223a2d.png
threadCount = 50

4.设置RunnerConnection threadCount数为100

img_113499062da504a0a523beb3886a95a9.png
threadCount = 50

可以看到随着 runnerConnection 连接线程数的递增,连接的稳定性是越来越低的。但用户调用不会长时间阻塞到connect fetch 上,而是按时返回。

目录
相关文章
|
19天前
|
安全 Java 程序员
深入理解Java内存模型与并发编程####
本文旨在探讨Java内存模型(JMM)的复杂性及其对并发编程的影响,不同于传统的摘要形式,本文将以一个实际案例为引子,逐步揭示JMM的核心概念,包括原子性、可见性、有序性,以及这些特性在多线程环境下的具体表现。通过对比分析不同并发工具类的应用,如synchronized、volatile关键字、Lock接口及其实现等,本文将展示如何在实践中有效利用JMM来设计高效且安全的并发程序。最后,还将简要介绍Java 8及更高版本中引入的新特性,如StampedLock,以及它们如何进一步优化多线程编程模型。 ####
21 0
|
21天前
|
Java 程序员
Java编程中的异常处理:从基础到高级
在Java的世界中,异常处理是代码健壮性的守护神。本文将带你从异常的基本概念出发,逐步深入到高级用法,探索如何优雅地处理程序中的错误和异常情况。通过实际案例,我们将一起学习如何编写更可靠、更易于维护的Java代码。准备好了吗?让我们一起踏上这段旅程,解锁Java异常处理的秘密!
|
1天前
|
存储 缓存 Java
Java 并发编程——volatile 关键字解析
本文介绍了Java线程中的`volatile`关键字及其与`synchronized`锁的区别。`volatile`保证了变量的可见性和一定的有序性,但不能保证原子性。它通过内存屏障实现,避免指令重排序,确保线程间数据一致。相比`synchronized`,`volatile`性能更优,适用于简单状态标记和某些特定场景,如单例模式中的双重检查锁定。文中还解释了Java内存模型的基本概念,包括主内存、工作内存及并发编程中的原子性、可见性和有序性。
Java 并发编程——volatile 关键字解析
|
5天前
|
算法 Java 调度
java并发编程中Monitor里的waitSet和EntryList都是做什么的
在Java并发编程中,Monitor内部包含两个重要队列:等待集(Wait Set)和入口列表(Entry List)。Wait Set用于线程的条件等待和协作,线程调用`wait()`后进入此集合,通过`notify()`或`notifyAll()`唤醒。Entry List则管理锁的竞争,未能获取锁的线程在此排队,等待锁释放后重新竞争。理解两者区别有助于设计高效的多线程程序。 - **Wait Set**:线程调用`wait()`后进入,等待条件满足被唤醒,需重新竞争锁。 - **Entry List**:多个线程竞争锁时,未获锁的线程在此排队,等待锁释放后获取锁继续执行。
32 12
|
1天前
|
存储 安全 Java
Java多线程编程秘籍:各种方案一网打尽,不要错过!
Java 中实现多线程的方式主要有四种:继承 Thread 类、实现 Runnable 接口、实现 Callable 接口和使用线程池。每种方式各有优缺点,适用于不同的场景。继承 Thread 类最简单,实现 Runnable 接口更灵活,Callable 接口支持返回结果,线程池则便于管理和复用线程。实际应用中可根据需求选择合适的方式。此外,还介绍了多线程相关的常见面试问题及答案,涵盖线程概念、线程安全、线程池等知识点。
11 2
|
24天前
|
缓存 Java 开发者
Java多线程编程的陷阱与最佳实践####
本文深入探讨了Java多线程编程中常见的陷阱,如竞态条件、死锁和内存一致性错误,并提供了实用的避免策略。通过分析典型错误案例,本文旨在帮助开发者更好地理解和掌握多线程环境下的编程技巧,从而提升并发程序的稳定性和性能。 ####
|
18天前
|
安全 算法 Java
Java多线程编程中的陷阱与最佳实践####
本文探讨了Java多线程编程中常见的陷阱,并介绍了如何通过最佳实践来避免这些问题。我们将从基础概念入手,逐步深入到具体的代码示例,帮助开发者更好地理解和应用多线程技术。无论是初学者还是有经验的开发者,都能从中获得有价值的见解和建议。 ####
|
18天前
|
Java 调度
Java中的多线程编程与并发控制
本文深入探讨了Java编程语言中多线程编程的基础知识和并发控制机制。文章首先介绍了多线程的基本概念,包括线程的定义、生命周期以及在Java中创建和管理线程的方法。接着,详细讲解了Java提供的同步机制,如synchronized关键字、wait()和notify()方法等,以及如何通过这些机制实现线程间的协调与通信。最后,本文还讨论了一些常见的并发问题,例如死锁、竞态条件等,并提供了相应的解决策略。
42 3
|
24天前
|
缓存 Java 开发者
Java多线程并发编程:同步机制与实践应用
本文深入探讨Java多线程中的同步机制,分析了多线程并发带来的数据不一致等问题,详细介绍了`synchronized`关键字、`ReentrantLock`显式锁及`ReentrantReadWriteLock`读写锁的应用,结合代码示例展示了如何有效解决竞态条件,提升程序性能与稳定性。
97 6
|
23天前
|
开发框架 安全 Java
Java 反射机制:动态编程的强大利器
Java反射机制允许程序在运行时检查类、接口、字段和方法的信息,并能操作对象。它提供了一种动态编程的方式,使得代码更加灵活,能够适应未知的或变化的需求,是开发框架和库的重要工具。
36 2