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();
@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;
}
@Override
public PooledObject<MqttConnection> wrap(MqttConnection mqttConnection) {
logger.info("封装默认返回类型 {}", mqttConnection.toString());
return new DefaultPooledObject<>(mqttConnection);
}
@Override
public void destroyObject(PooledObject<MqttConnection> p) throws Exception {
logger.info("销毁对象 {}", p.getObject().getMqttClient());
super.destroyObject(p);
}
@Override
public boolean validateObject(PooledObject<MqttConnection> p) {
logger.info("校验对象是否可用 {}", p.getObject().getMqttClient());
return super.validateObject(p);
}
@Override
public void activateObject(PooledObject<MqttConnection> p) throws Exception {
logger.info("激活钝化的对象 {}", p.getObject().getMqttClient());
super.activateObject(p);
}
@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 BasePooledObjectFactory 对象池化技术的文章就介绍到这了,更多相关Java 对象池内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！
