leveled queue

package com.leveldb.demo.service; import org.iq80.leveldb.*; import org.iq80.leveldb.impl.Iq80DBFactory; import org.springframework.util.StringUtils; import java.io.File; import java.io.IOException; public class LocalQueue { private static final String DB_PATH = "./database"; // LevelDB数据库路径 // private static DB db = null; public synchronized static String get(String key, DB db) { if (StringUtils.isEmpty(key)) { return null; } try { // byte[] r = db.get(JniDBFactory.bytes(key), new ReadOptions().fillCache(true).verifyChecksums(true)); // // return JniDBFactory.asString(r); byte[] r = db.get(Iq80DBFactory.bytes(key),new ReadOptions().fillCache(true).verifyChecksums(true)); return Iq80DBFactory.asString(r); } catch (Exception e) { e.printStackTrace(); } return null; } public synchronized static boolean put(String key, String data, DB db) { try { // db.put(JniDBFactory.bytes(key), JniDBFactory.bytes(data), new WriteOptions().sync(true)); db.put(Iq80DBFactory.bytes(key),Iq80DBFactory.bytes(data),new WriteOptions().sync(true)); return true; } catch (Exception e) { e.printStackTrace(); } return false; } public void close(DB db) { try { db.close(); } catch (IOException e) { e.printStackTrace(); } } /** * 压缩数据 */ public synchronized static void compactData(DB db) { try { //全部压缩数据 已经删除的数据，从硬盘上清除； db.compactRange(null, null); } catch (Exception e) { e.printStackTrace(); } } public synchronized static boolean delete(String key, DB db) { if (StringUtils.isEmpty(key)) { return false; } try { db.delete(Iq80DBFactory.bytes(key)); return true; } catch (Exception e) { e.printStackTrace(); } return false; } public void destroy(DB db) { close(db); } // public String getPath() { // return path; // } // // public void setPath(String path) { // this.path = path; // } public void enqueue(String item) throws Exception { Options options = new Options(); options.createIfMissing(true); DBFactory factory = new Iq80DBFactory(); try (DB db = factory.open(new File(DB_PATH), options)) { byte[] keyBytes = "next".getBytes("UTF-8"); String key = "next"; // 获取当前队尾索引值 byte[] valueBytes; // if (!get(keyBytes, db)) { // valueBytes = db.get(keyBytes); String value = get(key, db); if (StringUtils.isEmpty(value)) { value = "1"; } System.out.println("value is "+value); int nextIndex = Integer.parseInt(value); // 将新元素存入队列 byte[] dataKey = ("item" + nextIndex).getBytes("UTF-8"); System.out.println("item is "+item); byte[] dataValue = item.getBytes("UTF-8"); WriteOptions writeOpts = new WriteOptions(); writeOpts.sync(false); db.put(dataKey, dataValue); // 更新队尾索引值 nextIndex++; byte[] updatedNextIndex = String.valueOf(nextIndex).getBytes("UTF-8"); db.put(keyBytes, updatedNextIndex); } catch (Exception e) { throw e; } finally { factory.destroy(new File(DB_PATH), options); } } public static void main(String[] args) throws Exception { LocalQueue localQueue = new LocalQueue(); localQueue.enqueue("hello world"); System.out.println("putput is "+localQueue.dequeue()); } public String dequeue() throws Exception { Options options = new Options(); options.createIfMissing(true); DBFactory factory = new Iq80DBFactory(); try (DB db = factory.open(new File(DB_PATH), options)) { byte[] keyBytes = "next".getBytes("UTF-8"); // 获取当前队头索引值 // byte[] valueBytes = db.get(keyBytes); // int headIndex = Integer.parseInt(new String(valueBytes)); int headIndex = 1; // 从队列中移除并返回队头元素 System.out.println("head is "+headIndex); byte[] dataKey = ("item" + headIndex).getBytes("UTF-8"); byte[] dataValue = db.get(dataKey); db.delete(dataKey); return new String(dataValue); } catch (Exception e) { throw e; } finally { factory.destroy(new File(DB_PATH), options); } } }