Java GenericObjectPool 对象池化技术--SpringBoot sftp 连接池工具类
一个对象池包含一组已经初始化过且可以使用的对象，而可以在有需求时创建和销毁对象。池的用户可以从池子中取得对象，对其进行操作处理，并在不需要时归还给池子而非直接销毁它。这是一种特殊的工厂对象。
BasePooledObjectFactory 对象池化技术 的使用
Pom.xml
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
<version>2.7.0</version>
</dependency>
MqttConnection.java
package com.vipsoft.mqtt.pool;
public class MqttConnection {
private String mqttClient;
;
public MqttConnection(String mqttClient) {
this.mqttClient = mqttClient;
}
public String getMqttClient() {
return mqttClient;
}
public void setMqttClient(String mqttClient) {
this.mqttClient = mqttClient;
}
/**
* 推送方法消息
*/
public void publish(String msg) throws Exception {
System.out.println("对象" + mqttClient + ":" + "执行任务" + msg);
}
@Override
public String toString() {
return "MqttConnection{" + "id=" + mqttClient + '}';
}
}
MqttConnectionFactory.java
package com.vipsoft.mqtt.pool;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.atomic.AtomicInteger;
public class MqttConnectionFactory extends BasePooledObjectFactory<MqttConnection> {
private static final Logger logger = LoggerFactory.getLogger(MqttConnectionFactory.class);
// AtomicInteger是一个提供原子操作的Integer类，通过线程安全的方式操作加减
private AtomicInteger counter = new AtomicInteger();
/**
* 在对象池中创建对象
*
* @return
* @throws Exception
*/
@Override
public MqttConnection create() throws Exception {
// 实现线程安全避免在高并发的场景下出现clientId重复导致无法创建连接的情况
int count = this.counter.addAndGet(1);
MqttConnection mqttConnection = new MqttConnection("MqttConnection：" + count);
logger.info("在对象池中创建对象 {}", mqttConnection.toString());
return mqttConnection;
}
/**
* common-pool2 中创建了 DefaultPooledObject 对象对对象池中对象进行的包装。
* 将我们自定义的对象放置到这个包装中，工具会统计对象的状态、创建时间、更新时间、返回时间、出借时间、使用时间等等信息进行统计
*
* @param mqttConnection
* @return
*/
@Override
public PooledObject<MqttConnection> wrap(MqttConnection mqttConnection) {
logger.info("封装默认返回类型 {}", mqttConnection.toString());
return new DefaultPooledObject<>(mqttConnection);
}
/**
* 销毁对象
*
* @param p 对象池
* @throws Exception 异常
*/
@Override
public void destroyObject(PooledObject<MqttConnection> p) throws Exception {
logger.info("销毁对象 {}", p.getObject().getMqttClient());
super.destroyObject(p);
}
/**
* 校验对象是否可用
*
* @param p 对象池
* @return 对象是否可用结果，boolean
*/
@Override
public boolean validateObject(PooledObject<MqttConnection> p) {
logger.info("校验对象是否可用 {}", p.getObject().getMqttClient());
return super.validateObject(p);
}
/**
* 激活钝化的对象系列操作
*
* @param p 对象池
* @throws Exception 异常信息
*/
@Override
public void activateObject(PooledObject<MqttConnection> p) throws Exception {
logger.info("激活钝化的对象 {}", p.getObject().getMqttClient());
super.activateObject(p);
}
/**
* 钝化未使用的对象
*
* @param p 对象池
* @throws Exception 异常信息
*/
@Override
public void passivateObject(PooledObject<MqttConnection> p) throws Exception {
logger.info("钝化未使用的对象 {}", p.getObject().getMqttClient());
super.passivateObject(p);
}
}
PoolTest.java
package com.vipsoft.mqtt;
import cn.hutool.core.date.DateUtil;
import com.vipsoft.mqtt.pool.MqttConnection;
import com.vipsoft.mqtt.pool.MqttConnectionFactory;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
@SpringBootTest
public class PoolTest {
@Test
void basePooledTest() throws InterruptedException {
AtomicInteger atomicInteger = new AtomicInteger();
int excutorCount = 15;
CountDownLatch countDownLatch = new CountDownLatch(excutorCount);
// =====================创建线程池=====================
ExecutorService excutor = Executors.newFixedThreadPool(5);
// =====================创建对象池 项目中使用了InitializingBean.afterPropertiesSet() 中创建=====================
// 对象池工厂
MqttConnectionFactory personPoolFactory = new MqttConnectionFactory();
// 对象池配置
GenericObjectPoolConfig<MqttConnection> objectPoolConfig = new GenericObjectPoolConfig<>();
objectPoolConfig.setMaxTotal(50);
// 对象池
GenericObjectPool<MqttConnection> mqttPool = new GenericObjectPool<>(personPoolFactory, objectPoolConfig);
// =====================测试对象池=====================
// 循环100次，从线程池中取多个多线程执行任务，来测试对象池
for (int i = 0; i < excutorCount; i++) {
excutor.submit(new Thread(() -> {
// 模拟从对象池取出对象，执行任务
MqttConnection mqtt = null;
try {
// 从对象池取出对象
mqtt = mqttPool.borrowObject();
// 让对象工作
int count = atomicInteger.addAndGet(1);
mqtt.publish("Id:" + count + " Time: " + DateUtil.now());
} catch (Exception e) {
e.printStackTrace();
} finally {
// 回收对象到对象池
if (mqtt != null) {
mqttPool.returnObject(mqtt);
}
countDownLatch.countDown();
}
}));
}
countDownLatch.await();
}
}
本文简要介绍了Java 方法中循环调用具有事务的具体方法示例，虽然@Transactional是Spring中最常用和推荐的方式，但是本文还简要介绍了其他5种方法可以实现类似的功能。