NumberFormatter: use ThreadLocal for thread safety

DataDog · Mar 19, 2020 · 6f859d9 · 6f859d9
1 parent 888d97a
commit 6f859d9
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 11 deletions.
diff --git a/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java b/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java
@@ -82,15 +82,23 @@ public class NonBlockingStatsDClient implements StatsDClient {
      * The NumberFormat instances are not threadsafe but are only ever called from
      * the sender thread.
      */
-    private static final NumberFormat NUMBER_FORMATTER = newFormatter();
-    private static final NumberFormat SAMPLE_RATE_FORMATTER = newFormatter();
+    private static final ThreadLocal<NumberFormat> NUMBER_FORMATTER = new ThreadLocal<NumberFormat>() {
+        @Override
+        protected NumberFormat initialValue() {
+            return newFormatter(false);
+        }
+    };
+    private static final ThreadLocal<NumberFormat> SAMPLE_RATE_FORMATTER = new ThreadLocal<NumberFormat>() {
+        @Override
+        protected NumberFormat initialValue() {
+            return newFormatter(true);
+        }
+    };
 
     static {
-        NUMBER_FORMATTER.setMaximumFractionDigits(6);
-        SAMPLE_RATE_FORMATTER.setMinimumFractionDigits(6);
     }
 
-    private static NumberFormat newFormatter() {
+    private static NumberFormat newFormatter(boolean sampler) {
         // Always create the formatter for the US locale in order to avoid this bug:
         // https://github.com/indeedeng/java-dogstatsd-client/issues/3
         NumberFormat numberFormatter = NumberFormat.getInstance(Locale.US);
@@ -103,9 +111,20 @@ private static NumberFormat newFormatter() {
             symbols.setNaN("NaN");
             decimalFormat.setDecimalFormatSymbols(symbols);
         }
+
+        if (sampler) {
+            numberFormatter.setMinimumFractionDigits(6);
+        } else {
+            numberFormatter.setMaximumFractionDigits(6);
+        }
+
         return numberFormatter;
     }
 
+    private static String format(ThreadLocal<NumberFormat> formatter, double value) {
+        return formatter.get().format(value);
+    }
+
     private final String prefix;
     private final DatagramChannel clientChannel;
     private final StatsDClientErrorHandler handler;
@@ -353,7 +372,7 @@ public final void writeTo(StringBuilder builder) {
             writeValue(builder);
             builder.append('|').append(type);
             if (!Double.isNaN(sampleRate)) {
-                builder.append('|').append('@').append(SAMPLE_RATE_FORMATTER.format(sampleRate));
+                builder.append('|').append('@').append(format(SAMPLE_RATE_FORMATTER, sampleRate));
             }
             tagString(tags, builder);
         }
@@ -367,7 +386,7 @@ private void send(String aspect, final double value, String type, double sampleR
 
             statsDProcessor.send(new StatsDMessage(aspect, type, sampleRate, tags) {
                 @Override protected void writeValue(StringBuilder builder) {
-                    builder.append(NUMBER_FORMATTER.format(value));
+                    builder.append(format(NUMBER_FORMATTER, value));
                 }
             });
         }

diff --git a/src/main/java/com/timgroup/statsd/StatsDBlockingProcessor.java b/src/main/java/com/timgroup/statsd/StatsDBlockingProcessor.java
@@ -36,7 +36,6 @@ public void run() {
                     }
 
                     final Message message = messages.poll(WAIT_SLEEP_MS, TimeUnit.MILLISECONDS);
-                    // TODO: revisit this logic - cleanup, remove duplicate code.
                     if (message != null) {
 
                         builder.setLength(0);
@@ -66,7 +65,6 @@ public void run() {
                             writeBuilderToSendBuffer(sendBuffer);
                         }
 
-                        // TODO: revisit this logic
                         if (null == messages.peek()) {
                             outboundQueue.put(sendBuffer);
                             sendBuffer = bufferPool.borrow();

diff --git a/src/main/java/com/timgroup/statsd/StatsDNonBlockingProcessor.java b/src/main/java/com/timgroup/statsd/StatsDNonBlockingProcessor.java
@@ -41,7 +41,6 @@ public void run() {
                         return;
                     }
                     final Message message = messages.poll();
-                    // TODO: revisit this logic - cleanup, remove duplicate code.
                     if (message != null) {
 
                         qsize.decrementAndGet();
@@ -72,7 +71,6 @@ public void run() {
                             writeBuilderToSendBuffer(sendBuffer);
                         }
 
-                        // TODO: revisit this logic
                         if (null == messages.peek()) {
                             outboundQueue.put(sendBuffer);
                             sendBuffer = bufferPool.borrow();