wz2cool / local-queue / #5

Committed 27 Jan 2025 01:17PM UTC coverage: 92.442%. First build

Build # #5

Build Type

push

Committed by

Commit Message

Merge pull request #3 from wz2cool/0.0.5

0.0.5

Run Details

71 of 82 new or added lines in 5 files covered. (86.59%)

318 of 344 relevant lines covered (92.44%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.64

/src/main/java/com/github/wz2cool/localqueue/impl/SimpleProducer.java

package com.github.wz2cool.localqueue.impl;

import com.github.wz2cool.localqueue.IProducer;
import com.github.wz2cool.localqueue.model.config.SimpleProducerConfig;
import net.openhft.chronicle.queue.ChronicleQueue;
import net.openhft.chronicle.queue.ExcerptAppender;
import net.openhft.chronicle.queue.RollCycles;
import net.openhft.chronicle.queue.impl.single.SingleChronicleQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * simple writer
 *
 * @author frank
 */
public class SimpleProducer implements IProducer, AutoCloseable {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private final SimpleProducerConfig config;
    private final SingleChronicleQueue queue;
    private final LinkedBlockingQueue<String> messageCache = new LinkedBlockingQueue<>();
    private final ThreadLocal<ExcerptAppender> appenderThreadLocal;
    private final ExecutorService flushExecutor = Executors.newSingleThreadExecutor();
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    private final DateTimeFormatter dateFormatter = DateTimeFormatter.ofPattern("yyyyMMdd");
    private final Lock internalLock = new ReentrantLock();
    private volatile boolean isFlushRunning = true;
    private volatile boolean isClosing = false;
    private volatile boolean isClosed = false;

    public SimpleProducer(final SimpleProducerConfig config) {
        this.config = config;
        this.queue = ChronicleQueue.singleBuilder(config.getDataDir()).rollCycle(RollCycles.FAST_DAILY).build();
        this.appenderThreadLocal = ThreadLocal.withInitial(this.queue::createAppender);
        flushExecutor.execute(this::flush);
        scheduler.scheduleAtFixedRate(() -> cleanUpOldFiles(config.getKeepDays()), 0, 1, TimeUnit.HOURS);
    }


    /// region flush to file
    private void flush() {
        while (isFlushRunning && !isClosing) {
            flushInternal(config.getFlushBatchSize());
        }
    }

    private void stopFlush() {
        isFlushRunning = false;
    }

    private final List<String> tempFlushMessages = new ArrayList<>();

    private void flushInternal(int batchSize) {
        try {
            if (tempFlushMessages.isEmpty()) {
                // take 主要作用就是卡主线程
                String firstItem = this.messageCache.poll(config.getFlushInterval(), TimeUnit.MILLISECONDS);
                if (firstItem == null) {
                    return;
                }
                this.tempFlushMessages.add(firstItem);
                // 如果空了从消息缓存放入待刷消息
                this.messageCache.drainTo(tempFlushMessages, batchSize - 1);
            }
            flushInternal(tempFlushMessages);
            tempFlushMessages.clear();
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
        }
    }

    private void flushInternal(List<String> messages) {
        try {
            internalLock.lock();
            if (!isFlushRunning || isClosing) {
                return;
            }
            ExcerptAppender appender = appenderThreadLocal.get();
            for (String message : messages) {
                appender.writeText(message);
            }
        } finally {
            internalLock.unlock();
        }
    }

    /// endregion


    @Override
    public boolean offer(String message) {
        return this.messageCache.offer(message);
    }

    /**
     * get the last position
     *
     * @return this last position
     */
    public long getLastPosition() {
        return this.queue.lastIndex();
    }

    /// region close

    /**
     * is closed.
     *
     * @return true if the Producer is closed, false otherwise.
     */
    public boolean isClosed() {
        return isClosed;
    }

    @Override
    public void close() {
        isClosing = true;
        internalLock.lock();
        stopFlush();
        internalLock.unlock();
        queue.close();
        flushExecutor.shutdown();
        scheduler.shutdown();
        try {
            if (!scheduler.awaitTermination(1, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
            if (!flushExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
                flushExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            scheduler.shutdownNow();
            flushExecutor.shutdownNow();
            Thread.currentThread().interrupt();
        }

        appenderThreadLocal.remove();
        isClosed = true;
    }

    private void cleanUpOldFiles(int keepDays) {
        if (keepDays == -1) {
            // no need clean up old files
            return;
        }
        logger.debug("[cleanUpOldFiles] start");
        try {
            // Assuming .cq4 is the file extension for Chronicle Queue
            File[] files = config.getDataDir().listFiles((dir, name) -> name.endsWith(".cq4"));
            if (files == null || files.length == 0) {
                logger.debug("[cleanUpOldFiles] no files found");
                return;
            }
            LocalDate now = LocalDate.now();
            LocalDate keepStartDate = now.minusDays(keepDays);
            for (File file : files) {
                cleanUpOldFile(file, keepStartDate);
            }
        } catch (Exception ex) {
            logger.error("[cleanUpOldFiles] error", ex);
        } finally {
            logger.debug("[cleanUpOldFiles] end");
        }
    }

    private void cleanUpOldFile(final File file, final LocalDate keepDate) throws IOException {
        String fileName = file.getName();
        String dateString = fileName.substring(0, 8);
        LocalDate localDate = LocalDate.parse(dateString, this.dateFormatter);
        if (localDate.isBefore(keepDate)) {
            Files.deleteIfExists(file.toPath());
            logger.debug("[cleanUpOldFile] Deleted old file: {}", file.getName());
        }
    }

    /// endregion
}

1	package com.github.wz2cool.localqueue.impl;
2
3	import com.github.wz2cool.localqueue.IProducer;
4	import com.github.wz2cool.localqueue.model.config.SimpleProducerConfig;
5	import net.openhft.chronicle.queue.ChronicleQueue;
6	import net.openhft.chronicle.queue.ExcerptAppender;
7	import net.openhft.chronicle.queue.RollCycles;
8	import net.openhft.chronicle.queue.impl.single.SingleChronicleQueue;
9	import org.slf4j.Logger;
10	import org.slf4j.LoggerFactory;
11
12	import java.io.File;
13	import java.io.IOException;
14	import java.nio.file.Files;
15	import java.time.LocalDate;
16	import java.time.format.DateTimeFormatter;
17	import java.util.ArrayList;
18	import java.util.List;
19	import java.util.concurrent.*;
20	import java.util.concurrent.locks.Lock;
21	import java.util.concurrent.locks.ReentrantLock;
22
23	/**
24	* simple writer
25	*
26	* @author frank
27	*/
28	public class SimpleProducer implements IProducer, AutoCloseable {
29
30	private final Logger logger = LoggerFactory.getLogger(this.getClass());	1✔
31
32	private final SimpleProducerConfig config;
33	private final SingleChronicleQueue queue;
34	private final LinkedBlockingQueue<String> messageCache = new LinkedBlockingQueue<>();	1✔
35	private final ThreadLocal<ExcerptAppender> appenderThreadLocal;
36	private final ExecutorService flushExecutor = Executors.newSingleThreadExecutor();	1✔
37	private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);	1✔
38	private final DateTimeFormatter dateFormatter = DateTimeFormatter.ofPattern("yyyyMMdd");	1✔
39	private final Lock internalLock = new ReentrantLock();	1✔
40	private volatile boolean isFlushRunning = true;	1✔
41	private volatile boolean isClosing = false;	1✔
42	private volatile boolean isClosed = false;	1✔
43
44	public SimpleProducer(final SimpleProducerConfig config) {	1✔
45	this.config = config;	1✔
46	this.queue = ChronicleQueue.singleBuilder(config.getDataDir()).rollCycle(RollCycles.FAST_DAILY).build();	1✔
47	this.appenderThreadLocal = ThreadLocal.withInitial(this.queue::createAppender);	1✔
48	flushExecutor.execute(this::flush);	1✔
49	scheduler.scheduleAtFixedRate(() -> cleanUpOldFiles(config.getKeepDays()), 0, 1, TimeUnit.HOURS);	1✔
50	}	1✔
51
52
53	/// region flush to file
54	private void flush() {
55	while (isFlushRunning && !isClosing) {	1✔
56	flushInternal(config.getFlushBatchSize());	1✔
57	}
58	}	1✔
59
60	private void stopFlush() {
61	isFlushRunning = false;	1✔
62	}	1✔
63
64	private final List<String> tempFlushMessages = new ArrayList<>();	1✔
65
66	private void flushInternal(int batchSize) {
67	try {
68	if (tempFlushMessages.isEmpty()) {	1✔
69	// take 主要作用就是卡主线程
70	String firstItem = this.messageCache.poll(config.getFlushInterval(), TimeUnit.MILLISECONDS);	1✔
71	if (firstItem == null) {	1✔
72	return;	1✔
73	}
74	this.tempFlushMessages.add(firstItem);	1✔
75	// 如果空了从消息缓存放入待刷消息
76	this.messageCache.drainTo(tempFlushMessages, batchSize - 1);	1✔
77	}
78	flushInternal(tempFlushMessages);	1✔
79	tempFlushMessages.clear();	1✔
80	} catch (InterruptedException ex) {	×
81	Thread.currentThread().interrupt();	×
82	}	1✔
83	}	1✔
84
85	private void flushInternal(List<String> messages) {
86	try {
87	internalLock.lock();	1✔
88	if (!isFlushRunning \|\| isClosing) {	1✔
89	return;	×
90	}
91	ExcerptAppender appender = appenderThreadLocal.get();	1✔
92	for (String message : messages) {	1✔
93	appender.writeText(message);	1✔
94	}	1✔
95	} finally {
96	internalLock.unlock();	1✔
97	}	1✔
98	}	1✔
99
100	/// endregion
101
102
103	@Override
104	public boolean offer(String message) {
105	return this.messageCache.offer(message);	1✔
106	}
107
108	/**
109	* get the last position
110	*
111	* @return this last position
112	*/
113	public long getLastPosition() {
114	return this.queue.lastIndex();	1✔
115	}
116
117	/// region close
118
119	/**
120	* is closed.
121	*
122	* @return true if the Producer is closed, false otherwise.
123	*/
124	public boolean isClosed() {
125	return isClosed;	1✔
126	}
127
128	@Override
129	public void close() {
130	isClosing = true;	1✔
131	internalLock.lock();	1✔
132	stopFlush();	1✔
133	internalLock.unlock();	1✔
134	queue.close();	1✔
135	flushExecutor.shutdown();	1✔
136	scheduler.shutdown();	1✔
137	try {
138	if (!scheduler.awaitTermination(1, TimeUnit.SECONDS)) {	1✔
139	scheduler.shutdownNow();	×
140	}
141	if (!flushExecutor.awaitTermination(1, TimeUnit.SECONDS)) {	1✔
142	flushExecutor.shutdownNow();	×
143	}
NEW 144	} catch (InterruptedException e) {	×
NEW 145	scheduler.shutdownNow();	×
NEW 146	flushExecutor.shutdownNow();	×
NEW 147	Thread.currentThread().interrupt();	×
148	}	1✔
149
150	appenderThreadLocal.remove();	1✔
151	isClosed = true;	1✔
152	}	1✔
153
154	private void cleanUpOldFiles(int keepDays) {
155	if (keepDays == -1) {	1✔
156	// no need clean up old files
157	return;	1✔
158	}
159	logger.debug("[cleanUpOldFiles] start");	1✔
160	try {
161	// Assuming .cq4 is the file extension for Chronicle Queue
162	File[] files = config.getDataDir().listFiles((dir, name) -> name.endsWith(".cq4"));	1✔
163	if (files == null \|\| files.length == 0) {	1✔
164	logger.debug("[cleanUpOldFiles] no files found");	1✔
165	return;	1✔
166	}
167	LocalDate now = LocalDate.now();	1✔
168	LocalDate keepStartDate = now.minusDays(keepDays);	1✔
169	for (File file : files) {	1✔
170	cleanUpOldFile(file, keepStartDate);	1✔
171	}
172	} catch (Exception ex) {	×
173	logger.error("[cleanUpOldFiles] error", ex);	×
174	} finally {
175	logger.debug("[cleanUpOldFiles] end");	1✔
176	}	1✔
177	}	1✔
178
179	private void cleanUpOldFile(final File file, final LocalDate keepDate) throws IOException {
180	String fileName = file.getName();	1✔
181	String dateString = fileName.substring(0, 8);	1✔
182	LocalDate localDate = LocalDate.parse(dateString, this.dateFormatter);	1✔
183	if (localDate.isBefore(keepDate)) {	1✔
184	Files.deleteIfExists(file.toPath());	1✔
185	logger.debug("[cleanUpOldFile] Deleted old file: {}", file.getName());	1✔
186	}
187	}	1✔
188
189	/// endregion
190	}

wz2cool / local-queue / #5

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous