File tree Expand file tree Collapse file tree 3 files changed +20
-10
lines changed
src/main/java/com/timgroup/statsd Expand file tree Collapse file tree 3 files changed +20
-10
lines changed Original file line number Diff line number Diff line change @@ -112,11 +112,9 @@ protected ProcessingTask createProcessingTask() {
112
112
@ Override
113
113
protected boolean send (final Message message ) {
114
114
try {
115
- long threadId = Thread .currentThread ().getId ();
116
- // modulo reduction alternative to: long shard = threadID % this.lockShardGrain;
117
- // ref: https://lemire.me/blog/2016/06/27/a-fast-alternative-to-the-modulo-reduction/
118
- int shard = (int )((threadId * (long )this .lockShardGrain ) >> 32 );
119
- int processQueue = (int )((threadId * (long )this .workers ) >> 32 );
115
+ int threadId = getThreadId ();
116
+ int shard = threadId % lockShardGrain ;
117
+ int processQueue = threadId % workers ;
120
118
121
119
if (!shutdown ) {
122
120
messages [shard ].put (message );
Original file line number Diff line number Diff line change @@ -186,11 +186,9 @@ protected ProcessingTask createProcessingTask() {
186
186
@ Override
187
187
protected boolean send (final Message message ) {
188
188
if (!shutdown ) {
189
- long threadId = Thread .currentThread ().getId ();
190
- // modulo reduction alternative to: long shard = threadID % [shard]this.lockShardGrain;
191
- // ref: https://lemire.me/blog/2016/06/27/a-fast-alternative-to-the-modulo-reduction/
192
- int shard = (int )((threadId * (long )this .lockShardGrain ) >> 32 );
193
- int processQueue = (int )((threadId * (long )this .workers ) >> 32 );
189
+ int threadId = getThreadId ();
190
+ int shard = threadId % lockShardGrain ;
191
+ int processQueue = threadId % workers ;
194
192
195
193
if (qsize [shard ].get () < qcapacity ) {
196
194
messages [shard ].offer (message );
Original file line number Diff line number Diff line change @@ -27,12 +27,21 @@ public abstract class StatsDProcessor implements Runnable {
27
27
protected static final String MESSAGE_TOO_LONG = "Message longer than size of sendBuffer" ;
28
28
protected static final int WAIT_SLEEP_MS = 10 ; // 10 ms would be a 100HZ slice
29
29
30
+ // Atomic integer containing the next thread ID to be assigned
31
+ private static final AtomicInteger nextId = new AtomicInteger (0 );
32
+
30
33
protected final StatsDClientErrorHandler handler ;
31
34
32
35
protected final BufferPool bufferPool ;
33
36
protected final BlockingQueue <ByteBuffer > outboundQueue ; // FIFO queue with outbound buffers
34
37
protected final ExecutorService executor ;
35
38
protected final CountDownLatch endSignal ;
39
+ protected static final ThreadLocal <Integer > threadId = new ThreadLocal <Integer >() {
40
+ @ Override
41
+ protected Integer initialValue () {
42
+ return nextId .getAndIncrement ();
43
+ }
44
+ };
36
45
37
46
protected final int workers ;
38
47
protected final int lockShardGrain ;
@@ -97,6 +106,11 @@ public BlockingQueue<ByteBuffer> getOutboundQueue() {
97
106
return this .outboundQueue ;
98
107
}
99
108
109
+ // Returns the current thread's unique ID, assigning it if necessary
110
+ public static int getThreadId () {
111
+ return threadId .get ().intValue ();
112
+ }
113
+
100
114
@ Override
101
115
public void run () {
102
116
You can’t perform that action at this time.
0 commit comments