wz2cool / local-queue / #94

Committed 06 Jul 2025 07:25AM UTC coverage: 85.65% (-5.2%) from 90.82%

Build # #94

Build Type

push

Committed by

Commit Message

add option

Coverage Stats

4 of 20 new or added lines in 4 files covered. (20.0%)

758 of 885 relevant lines covered (85.65%)

0.86 hits per line

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

81.16

/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.event.CloseListener;
import com.github.wz2cool.localqueue.helper.ChronicleQueueHelper;
import com.github.wz2cool.localqueue.model.config.SimpleProducerConfig;
import com.github.wz2cool.localqueue.model.message.MessageOption;
import com.github.wz2cool.localqueue.model.message.internal.HeaderMessage;
import com.github.wz2cool.localqueue.model.message.internal.InternalMessage;
import net.openhft.chronicle.core.time.TimeProvider;
import net.openhft.chronicle.queue.ChronicleQueue;
import net.openhft.chronicle.queue.ExcerptAppender;
import net.openhft.chronicle.queue.RollCycle;
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.atomic.AtomicBoolean;

/**
 * simple writer
 *
 * @author frank
 */
@SuppressWarnings("Duplicates")
public class SimpleProducer implements IProducer {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    private final RollCycle defaultRollCycle;
    private final TimeProvider timeProvider;
    private final SimpleProducerConfig config;
    private final SingleChronicleQueue queue;
    private final LinkedBlockingQueue<InternalMessage> messageCache = new LinkedBlockingQueue<>();
    // should only call by flushExecutor
    private final ExcerptAppender mainAppender;
    private final ExecutorService flushExecutor = Executors.newSingleThreadExecutor();
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    private final DateTimeFormatter dateFormatter = DateTimeFormatter.ofPattern("yyyyMMdd");
    private final ConcurrentLinkedQueue<CloseListener> closeListeners = new ConcurrentLinkedQueue<>();

    private final AtomicBoolean isFlushRunning = new AtomicBoolean(true);
    private final AtomicBoolean isClosing = new AtomicBoolean(false);
    private final AtomicBoolean isClosed = new AtomicBoolean(false);
    private final Object closeLocker = new Object();

    public SimpleProducer(final SimpleProducerConfig config) {
        this.config = config;
        this.timeProvider = ChronicleQueueHelper.getTimeProvider(config.getTimeZone());
        this.defaultRollCycle = ChronicleQueueHelper.getRollCycle(config.getRollCycleType());
        this.queue = ChronicleQueue.singleBuilder(config.getDataDir())
                .rollCycle(defaultRollCycle)
                .timeProvider(timeProvider)
                .build();
        this.mainAppender = initMainAppender();
        flushExecutor.execute(this::flush);
        scheduler.scheduleAtFixedRate(() -> cleanUpOldFiles(config.getKeepDays()), 0, 1, TimeUnit.HOURS);
    }

    private ExcerptAppender initMainAppender() {
        return CompletableFuture.supplyAsync(this.queue::createAppender, this.flushExecutor).join();
    }

    // region flush to file

    private void stopFlush() {
        isFlushRunning.set(false);
    }

    private void flush() {
        while (isFlushRunning.get() && !isClosing.get()) {
            flushMessages(config.getFlushBatchSize());
        }
    }

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

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

    private void doFlushMessages(final List<InternalMessage> messages) {
        synchronized (closeLocker) {
            try {
                logDebug("[flushMessages] start");
                if (isClosing.get()) {
                    logDebug("[flushMessages] producer is closing");
                    return;
                }

                for (InternalMessage message : messages) {
                    long writeTime = System.currentTimeMillis();
                    message.setWriteTime(writeTime);
                    mainAppender.writeBytes(message);
                }
            } finally {
                logDebug("[flushMessages] end");
            }
        }
    }

    // endregion


    @Override
    public boolean offer(String message) {
        return offer(null, message);
    }

    @Override
    public boolean offer(String messageKey, String message) {
        InternalMessage internalWriteMessage = new InternalMessage();
        internalWriteMessage.setContent(message);
        internalWriteMessage.setMessageKey(messageKey);
        return this.messageCache.offer(internalWriteMessage);
    }

    @Override
    public boolean offer(String tag, String messageKey, String message) {
        InternalMessage internalWriteMessage = new InternalMessage();
        internalWriteMessage.setContent(message);
        internalWriteMessage.setMessageKey(messageKey);
        internalWriteMessage.setTag(tag);
        return this.messageCache.offer(internalWriteMessage);
    }

    @Override
    public boolean offer(String message, MessageOption option) {
        InternalMessage internalWriteMessage = new InternalMessage();
        internalWriteMessage.setContent(message);
        internalWriteMessage.setMessageKey(option.getMessageKey());
        internalWriteMessage.setTag(option.getTag());
        if (option.hasHeader()) {
            HeaderMessage headerMessage = new HeaderMessage(option.getHeaders());
            internalWriteMessage.setHeaderMessage(headerMessage);
        }
        return this.messageCache.offer(internalWriteMessage);
    }

    /**
     * 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.
     */
    @Override
    public boolean isClosed() {
        return isClosed.get();
    }

    @Override
    public void close() {
        logDebug("[close] start");
        if (isClosing.get()) {
            logDebug("[close] is closing");
            return;
        }
        isClosing.set(true);
        stopFlush();
        synchronized (closeLocker) {
            try {
                if (!queue.isClosed()) {
                    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();
                }
                for (CloseListener closeListener : closeListeners) {
                    closeListener.onClose();
                }
                isClosed.set(true);
            } finally {
                logDebug("[close] end");
            }
        }
    }

    @Override
    public void addCloseListener(CloseListener listener) {
        closeListeners.add(listener);
    }

    private void cleanUpOldFiles(int keepDays) {
        if (keepDays == -1) {
            // no need clean up old files
            return;
        }
        logDebug("[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) {
                logDebug("[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 {
            logDebug("[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());
            logDebug("[cleanUpOldFile] Deleted old file: {}", file.getName());
        }
    }

    // endregion

    // region logger
    private void logDebug(String format, String arg) {
        if (logger.isDebugEnabled()) {
            logger.debug(format, arg);
        }
    }


    private void logDebug(String format) {
        if (logger.isDebugEnabled()) {
            logger.debug(format);
        }
    }

    // endregion
}

1	package com.github.wz2cool.localqueue.impl;
2
3	import com.github.wz2cool.localqueue.IProducer;
4	import com.github.wz2cool.localqueue.event.CloseListener;
5	import com.github.wz2cool.localqueue.helper.ChronicleQueueHelper;
6	import com.github.wz2cool.localqueue.model.config.SimpleProducerConfig;
7	import com.github.wz2cool.localqueue.model.message.MessageOption;
8	import com.github.wz2cool.localqueue.model.message.internal.HeaderMessage;
9	import com.github.wz2cool.localqueue.model.message.internal.InternalMessage;
10	import net.openhft.chronicle.core.time.TimeProvider;
11	import net.openhft.chronicle.queue.ChronicleQueue;
12	import net.openhft.chronicle.queue.ExcerptAppender;
13	import net.openhft.chronicle.queue.RollCycle;
14	import net.openhft.chronicle.queue.impl.single.SingleChronicleQueue;
15	import org.slf4j.Logger;
16	import org.slf4j.LoggerFactory;
17
18	import java.io.File;
19	import java.io.IOException;
20	import java.nio.file.Files;
21	import java.time.LocalDate;
22	import java.time.format.DateTimeFormatter;
23	import java.util.ArrayList;
24	import java.util.List;
25	import java.util.concurrent.*;
26	import java.util.concurrent.atomic.AtomicBoolean;
27
28	/**
29	* simple writer
30	*
31	* @author frank
32	*/
33	@SuppressWarnings("Duplicates")
34	public class SimpleProducer implements IProducer {
35
36	private final Logger logger = LoggerFactory.getLogger(this.getClass());	1✔
37	private final RollCycle defaultRollCycle;
38	private final TimeProvider timeProvider;
39	private final SimpleProducerConfig config;
40	private final SingleChronicleQueue queue;
41	private final LinkedBlockingQueue<InternalMessage> messageCache = new LinkedBlockingQueue<>();	1✔
42	// should only call by flushExecutor
43	private final ExcerptAppender mainAppender;
44	private final ExecutorService flushExecutor = Executors.newSingleThreadExecutor();	1✔
45	private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);	1✔
46	private final DateTimeFormatter dateFormatter = DateTimeFormatter.ofPattern("yyyyMMdd");	1✔
47	private final ConcurrentLinkedQueue<CloseListener> closeListeners = new ConcurrentLinkedQueue<>();	1✔
48
49	private final AtomicBoolean isFlushRunning = new AtomicBoolean(true);	1✔
50	private final AtomicBoolean isClosing = new AtomicBoolean(false);	1✔
51	private final AtomicBoolean isClosed = new AtomicBoolean(false);	1✔
52	private final Object closeLocker = new Object();	1✔
53
54	public SimpleProducer(final SimpleProducerConfig config) {	1✔
55	this.config = config;	1✔
56	this.timeProvider = ChronicleQueueHelper.getTimeProvider(config.getTimeZone());	1✔
57	this.defaultRollCycle = ChronicleQueueHelper.getRollCycle(config.getRollCycleType());	1✔
58	this.queue = ChronicleQueue.singleBuilder(config.getDataDir())	1✔
59	.rollCycle(defaultRollCycle)	1✔
60	.timeProvider(timeProvider)	1✔
61	.build();	1✔
62	this.mainAppender = initMainAppender();	1✔
63	flushExecutor.execute(this::flush);	1✔
64	scheduler.scheduleAtFixedRate(() -> cleanUpOldFiles(config.getKeepDays()), 0, 1, TimeUnit.HOURS);	1✔
65	}	1✔
66
67	private ExcerptAppender initMainAppender() {
68	return CompletableFuture.supplyAsync(this.queue::createAppender, this.flushExecutor).join();	1✔
69	}
70
71	// region flush to file
72
73	private void stopFlush() {
74	isFlushRunning.set(false);	1✔
75	}	1✔
76
77	private void flush() {
78	while (isFlushRunning.get() && !isClosing.get()) {	1✔
79	flushMessages(config.getFlushBatchSize());	1✔
80	}
81	}	1✔
82
83	private final List<InternalMessage> tempFlushMessages = new ArrayList<>();	1✔
84
85	private void flushMessages(int batchSize) {
86	try {
87	logDebug("[flushInternal] start");	1✔
88	if (tempFlushMessages.isEmpty()) {	1✔
89	// take 主要作用就是卡主线程
90	InternalMessage firstItem = this.messageCache.poll(config.getFlushInterval(), TimeUnit.MILLISECONDS);	1✔
91	if (firstItem == null) {	1✔
92	return;	1✔
93	}
94	this.tempFlushMessages.add(firstItem);	1✔
95	// 如果空了从消息缓存放入待刷消息
96	this.messageCache.drainTo(tempFlushMessages, batchSize - 1);	1✔
97	}
98	doFlushMessages(tempFlushMessages);	1✔
99	tempFlushMessages.clear();	1✔
100	} catch (InterruptedException ex) {	×
101	Thread.currentThread().interrupt();	×
102	} catch (Exception ex) {	×
103	logger.error("[flushInternal] flush error", ex);	×
104	} finally {
105	logDebug("[flushInternal] end");	1✔
106	}	1✔
107	}	1✔
108
109	private void doFlushMessages(final List<InternalMessage> messages) {
110	synchronized (closeLocker) {	1✔
111	try {
112	logDebug("[flushMessages] start");	1✔
113	if (isClosing.get()) {	1✔
114	logDebug("[flushMessages] producer is closing");	×
115	return;	×
116	}
117
118	for (InternalMessage message : messages) {	1✔
119	long writeTime = System.currentTimeMillis();	1✔
120	message.setWriteTime(writeTime);	1✔
121	mainAppender.writeBytes(message);	1✔
122	}	1✔
123	} finally {
124	logDebug("[flushMessages] end");	1✔
125	}	1✔
126	}	1✔
127	}	1✔
128
129	// endregion
130
131
132	@Override
133	public boolean offer(String message) {
134	return offer(null, message);	1✔
135	}
136
137	@Override
138	public boolean offer(String messageKey, String message) {
139	InternalMessage internalWriteMessage = new InternalMessage();	1✔
140	internalWriteMessage.setContent(message);	1✔
141	internalWriteMessage.setMessageKey(messageKey);	1✔
142	return this.messageCache.offer(internalWriteMessage);	1✔
143	}
144
145	@Override
146	public boolean offer(String tag, String messageKey, String message) {
147	InternalMessage internalWriteMessage = new InternalMessage();	1✔
148	internalWriteMessage.setContent(message);	1✔
149	internalWriteMessage.setMessageKey(messageKey);	1✔
150	internalWriteMessage.setTag(tag);	1✔
151	return this.messageCache.offer(internalWriteMessage);	1✔
152	}
153
154	@Override
155	public boolean offer(String message, MessageOption option) {
NEW 156	InternalMessage internalWriteMessage = new InternalMessage();	×
NEW 157	internalWriteMessage.setContent(message);	×
NEW 158	internalWriteMessage.setMessageKey(option.getMessageKey());	×
NEW 159	internalWriteMessage.setTag(option.getTag());	×
NEW 160	if (option.hasHeader()) {	×
NEW 161	HeaderMessage headerMessage = new HeaderMessage(option.getHeaders());	×
NEW 162	internalWriteMessage.setHeaderMessage(headerMessage);	×
163	}
NEW 164	return this.messageCache.offer(internalWriteMessage);	×
165	}
166
167	/**
168	* get the last position
169	*
170	* @return this last position
171	*/
172	public long getLastPosition() {
173	return this.queue.lastIndex();	1✔
174	}
175
176	// region close
177
178	/**
179	* is closed.
180	*
181	* @return true if the Producer is closed, false otherwise.
182	*/
183	@Override
184	public boolean isClosed() {
185	return isClosed.get();	1✔
186	}
187
188	@Override
189	public void close() {
190	logDebug("[close] start");	1✔
191	if (isClosing.get()) {	1✔
192	logDebug("[close] is closing");	×
193	return;	×
194	}
195	isClosing.set(true);	1✔
196	stopFlush();	1✔
197	synchronized (closeLocker) {	1✔
198	try {
199	if (!queue.isClosed()) {	1✔
200	queue.close();	1✔
201	}
202	flushExecutor.shutdown();	1✔
203	scheduler.shutdown();	1✔
204	try {
205	if (!scheduler.awaitTermination(1, TimeUnit.SECONDS)) {	1✔
206	scheduler.shutdownNow();	×
207	}
208	if (!flushExecutor.awaitTermination(1, TimeUnit.SECONDS)) {	1✔
209	flushExecutor.shutdownNow();	×
210	}
211	} catch (InterruptedException e) {	×
212	scheduler.shutdownNow();	×
213	flushExecutor.shutdownNow();	×
214	Thread.currentThread().interrupt();	×
215	}	1✔
216	for (CloseListener closeListener : closeListeners) {	1✔
217	closeListener.onClose();	1✔
218	}	1✔
219	isClosed.set(true);	1✔
220	} finally {
221	logDebug("[close] end");	1✔
222	}	1✔
223	}	1✔
224	}	1✔
225
226	@Override
227	public void addCloseListener(CloseListener listener) {
228	closeListeners.add(listener);	1✔
229	}	1✔
230
231	private void cleanUpOldFiles(int keepDays) {
232	if (keepDays == -1) {	1✔
233	// no need clean up old files
234	return;	1✔
235	}
236	logDebug("[cleanUpOldFiles] start");	1✔
237	try {
238	// Assuming .cq4 is the file extension for Chronicle Queue
239	File[] files = config.getDataDir().listFiles((dir, name) -> name.endsWith(".cq4"));	1✔
240	if (files == null \|\| files.length == 0) {	1✔
241	logDebug("[cleanUpOldFiles] no files found");	1✔
242	return;	1✔
243	}
244	LocalDate now = LocalDate.now();	1✔
245	LocalDate keepStartDate = now.minusDays(keepDays);	1✔
246	for (File file : files) {	1✔
247	cleanUpOldFile(file, keepStartDate);	1✔
248	}
249	} catch (Exception ex) {	×
250	logger.error("[cleanUpOldFiles] error", ex);	×
251	} finally {
252	logDebug("[cleanUpOldFiles] end");	1✔
253	}	1✔
254	}	1✔
255
256
257	private void cleanUpOldFile(final File file, final LocalDate keepDate) throws IOException {
258	String fileName = file.getName();	1✔
259	String dateString = fileName.substring(0, 8);	1✔
260	LocalDate localDate = LocalDate.parse(dateString, this.dateFormatter);	1✔
261	if (localDate.isBefore(keepDate)) {	1✔
262	Files.deleteIfExists(file.toPath());	1✔
263	logDebug("[cleanUpOldFile] Deleted old file: {}", file.getName());	1✔
264	}
265	}	1✔
266
267	// endregion
268
269	// region logger
270	private void logDebug(String format, String arg) {
271	if (logger.isDebugEnabled()) {	1✔
272	logger.debug(format, arg);	×
273	}
274	}	1✔
275
276
277	private void logDebug(String format) {
278	if (logger.isDebugEnabled()) {	1✔
279	logger.debug(format);	×
280	}
281	}	1✔
282
283	// endregion
284	}

wz2cool / local-queue / #94

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