Berkeley DB是一个开放源代码的内嵌式数据库管理系统，能够为应用程序提供高性能的数据管理服务。应用它程序员只需要调用一些简单的API就可以完成对数据的访问和管理。与常用的数据库管理系统（如MySQL和Oracle等）有所不同，在Berkeley DB中并没有数据库服务器的概念。应用程序不需要事先同数据库服务建立起网络连接，而是通过内嵌在程序中的Berkeley DB函数库来完成对数据的保存、查询、修改和删除等操作。
1. 引入maven依赖
<dependency>
<groupId>com.sleepycat</groupId>
<artifactId>je</artifactId>
<version>18.3.12</version>
</dependency>
<dependency>
<groupId>de.ruedigermoeller</groupId>
<artifactId>fst</artifactId>
<version>2.48-jdk-6</version>
<exclusions>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
<exclusion>
<groupId>com.cedarsoftware</groupId>
<artifactId>json-io</artifactId>
</exclusion>
</exclusions>
</dependency>
2. 具体实现
BdbInitService文件
package com.joe.heightconcurrency.demo.service;
import com.joe.heightconcurrency.demo.config.BdbPropertyConfig;
import com.joe.heightconcurrency.demo.entity.User;
import com.joe.heightconcurrency.demo.util.FstSerialBinding;
import com.sleepycat.collections.StoredMap;
import com.sleepycat.je.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.io.File;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
@Service
@Slf4j
public class BdbInitService {
private static Environment env;
private static DatabaseConfig dbConfig;
@Autowired
private BdbPropertyConfig bdbPropertyConfig;
private ConcurrentHashMap<String, Database> queueDbMap = new ConcurrentHashMap<>();
private ConcurrentHashMap<String, StoredMap<String, User>> storedMaps = new ConcurrentHashMap<>();
static {
EnvironmentConfig envConfig = new EnvironmentConfig();
envConfig.setAllowCreate(true);
envConfig.setTransactional(true);
envConfig.setCacheSize(10000000);
try {
env = new Environment(new File("E://develop//bdb"),envConfig);
} catch (DatabaseException e) {
e.printStackTrace();
}
dbConfig = new DatabaseConfig();
dbConfig.setAllowCreate(true);
dbConfig.setTransactional(true);
dbConfig.setSortedDuplicates(false);
}
private Database open(String dbName) {
Database database = queueDbMap.get(dbName);
if (Objects.nonNull(database)) {
return database;
}
try {
database = env.openDatabase(null, dbName, dbConfig);
Database olddb = queueDbMap.putIfAbsent(dbName, database);
return Objects.isNull(olddb) ? database : olddb;
} catch (DatabaseException e) {
e.printStackTrace();
return null;
}
}
public StoredMap<String, User> getStoredMap(String queueName) {
FstSerialBinding<String> messageKeyBinding = new FstSerialBinding<>();
FstSerialBinding<User> messageValueBinding = new FstSerialBinding<>();
Database database = open(queueName);
StoredMap<String, User> tmpMap = new StoredMap<>(database, messageKeyBinding, messageValueBinding, true);
StoredMap<String, User> oldMap = storedMaps.putIfAbsent(queueName, tmpMap);
return Objects.isNull(oldMap) ? tmpMap : oldMap;
}
}
UserQueueService文件
package com.joe.heightconcurrency.demo.service;
import com.joe.heightconcurrency.demo.entity.User;
import com.joe.heightconcurrency.demo.queue.BdbBlockingQueue;
import com.sleepycat.collections.StoredMap;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.util.Objects;
@Service
@Slf4j
public class UserQueueService {
BdbBlockingQueue<User> bdbBlockingQueue;
@Autowired
private BdbInitService bdbInitService;
@PostConstruct
public void init() {
String queueName = "user";
StoredMap<String, User> storedMap = bdbInitService.getStoredMap(queueName);
bdbBlockingQueue = new BdbBlockingQueue<>(storedMap);
}
public void addUser(User user) {
if (Objects.isNull(user)) {
log.error("用户信息不存在");
return;
}
boolean result = bdbBlockingQueue.offer(user);
if (result) {
log.info("用户信息放入队列成功");
log.info("当前队列大小：{}", bdbBlockingQueue.size());
} else {
log.error("用户信息放入队列失败");
log.info("当前队列大小：{}", bdbBlockingQueue.size());
}
}
public Integer getSize() {
return bdbBlockingQueue.size();
}
public User getUser() {
return bdbBlockingQueue.poll();
}
}
controller文件
@GetMapping("/addUser")
public String addUser(User user) {
userQueueService.addUser(user);
return "SUCCESS";
}
@GetMapping("/getUserSize")
public Integer getUserSize() {
return userQueueService.getSize();
}
@GetMapping("/getUser")
public String getUser() {
return userQueueService.getUser().toString();
}
FstSerialBinding文件
package com.joe.heightconcurrency.demo.util;
import com.sleepycat.bind.ByteArrayBinding;
import com.sleepycat.bind.EntryBinding;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.util.RuntimeExceptionWrapper;
import java.io.Serializable;
public class FstSerialBinding<E extends Serializable> implements EntryBinding<E> {
private final static byte[] ZERO_LENGTH_BYTE_ARRAY = new byte[0];
private ByteArrayBinding bb = new ByteArrayBinding();
@Override
public E entryToObject(DatabaseEntry entry) {
byte[] data = bb.entryToObject(entry);
if(data ==null || data.length ==0) return null;
try{
return (E) FstObjectSerializeUtil.read(data);
}catch(Exception ex){
throw RuntimeExceptionWrapper.wrapIfNeeded(ex);
}
}
@Override
public void objectToEntry(E object, DatabaseEntry entry) {
if(object==null){
bb.objectToEntry(ZERO_LENGTH_BYTE_ARRAY, entry);
}else{
try{
bb.objectToEntry(FstObjectSerializeUtil.write(object), entry);
}catch(Exception ex){
throw RuntimeExceptionWrapper.wrapIfNeeded(ex);
}
}
}
}
FstObjectSerializeUtil文件
package com.joe.heightconcurrency.demo.util;
import org.nustaq.serialization.FSTConfiguration;
import org.nustaq.serialization.FSTObjectInput;
import org.nustaq.serialization.FSTObjectOutput;
import java.io.ByteArrayOutputStream;
import java.io.Serializable;
public abstract class FstObjectSerializeUtil {
private final static ThreadLocal<FSTConfiguration> conf = new ThreadLocal<FSTConfiguration>() {
public FSTConfiguration initialValue() {
return FSTConfiguration.createDefaultConfiguration();
}
};
public static byte[] write(Serializable obj) throws Exception{
ByteArrayOutputStream arroutput = new ByteArrayOutputStream();
FSTObjectOutput objoutput = conf.get().getObjectOutput(arroutput);
try{
objoutput.writeObject(obj);
objoutput.flush();
return arroutput.toByteArray();
}finally{
arroutput.close();
}
}
public static Serializable read(byte[] bytes) throws Exception{
FSTObjectInput objinput = conf.get().getObjectInput(bytes);
try{
Object t = objinput.readObject();
if(t instanceof Serializable){
return (Serializable)t;
}else{
return null;
}
}catch(Exception ex) {
return null;
}finally{
}
}
}
BdbBlockingQueue文件
package com.joe.heightconcurrency.demo.queue;
import com.joe.heightconcurrency.demo.entity.User;
import com.sleepycat.collections.StoredMap;
import java.util.Objects;
import java.util.concurrent.LinkedBlockingQueue;
public class BdbBlockingQueue<T extends User> extends LinkedBlockingQueue<T> {
private StoredMap<String, T> storedMap;
public BdbBlockingQueue() {}
public BdbBlockingQueue(StoredMap<String, T> storedMap) {
super(storedMap.values());
this.storedMap = storedMap;
}
@Override
public boolean offer(T t) {
if (Objects.isNull(t)) {
return false;
}
storedMap.put(t.getId().toString(), t);
return super.offer(t);
}
@Override
public int size() {
return super.size();
}
@Override
public T poll() {
T t = super.poll();
if (Objects.nonNull(t)) {
storedMap.remove(t.getId().toString());
}
return t;
}
}
参考：https://blog.csdn.net/qq_38617531/article/details/83794893/
网友评论