clean release messages on demand

apollo-biz/src/main/java/com/ctrip/framework/apollo/biz/message/DatabaseMessageSender.java

 package com.ctrip.framework.apollo.biz.message;
 
+import com.google.common.collect.Queues;
+
 import com.ctrip.framework.apollo.biz.entity.ReleaseMessage;
 import com.ctrip.framework.apollo.biz.repository.ReleaseMessageRepository;
+import com.ctrip.framework.apollo.core.utils.ApolloThreadFactory;
 import com.ctrip.framework.apollo.tracer.Tracer;
 import com.ctrip.framework.apollo.tracer.spi.Transaction;
 
@@ -9,8 +12,17 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
 
+import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import javax.annotation.PostConstruct;
 
 /**
  * @author Jason Song(song_s@ctrip.com)
@@ -18,11 +30,21 @@ import java.util.Objects;
 @Component
 public class DatabaseMessageSender implements MessageSender {
   private static final Logger logger = LoggerFactory.getLogger(DatabaseMessageSender.class);
+  private static final int CLEAN_QUEUE_MAX_SIZE = 100;
+  private BlockingQueue<Long> toClean = Queues.newLinkedBlockingQueue(CLEAN_QUEUE_MAX_SIZE);
+  private final ExecutorService cleanExecutorService;
+  private final AtomicBoolean cleanStopped;
 
   @Autowired
   private ReleaseMessageRepository releaseMessageRepository;
 
+  public DatabaseMessageSender() {
+    cleanExecutorService = Executors.newSingleThreadExecutor(ApolloThreadFactory.create("DatabaseMessageSender", true));
+    cleanStopped = new AtomicBoolean(false);
+  }
+
   @Override
+  @Transactional
   public void sendMessage(String message, String channel) {
     logger.info("Sending message {} to channel {}", message, channel);
     if (!Objects.equals(channel, Topics.APOLLO_RELEASE_TOPIC)) {
@@ -33,7 +55,8 @@ public class DatabaseMessageSender implements MessageSender {
     Tracer.logEvent("Apollo.AdminService.ReleaseMessage", message);
     Transaction transaction = Tracer.newTransaction("Apollo.AdminService", "sendMessage");
     try {
-      releaseMessageRepository.save(new ReleaseMessage(message));
+      ReleaseMessage newMessage = releaseMessageRepository.save(new ReleaseMessage(message));
+      toClean.offer(newMessage.getId());
       transaction.setStatus(Transaction.SUCCESS);
     } catch (Throwable ex) {
       logger.error("Sending message to database failed", ex);
@@ -42,4 +65,45 @@ public class DatabaseMessageSender implements MessageSender {
       transaction.complete();
     }
   }
+
+  @PostConstruct
+  private void initialize() {
+    cleanExecutorService.submit(() -> {
+      while (!cleanStopped.get() && !Thread.currentThread().isInterrupted()) {
+        try {
+          Long rm = toClean.poll(1, TimeUnit.SECONDS);
+          if (rm != null) {
+            cleanMessage(rm);
+          } else {
+            TimeUnit.SECONDS.sleep(5);
+          }
+        } catch (Throwable ex) {
+          Tracer.logError(ex);
+        }
+      }
+    });
+  }
+
+  private void cleanMessage(Long id) {
+    boolean hasMore = true;
+    //double check in case the release message is rolled back
+    ReleaseMessage releaseMessage = releaseMessageRepository.findOne(id);
+    if (releaseMessage == null) {
+      return;
+    }
+    while (hasMore && !Thread.currentThread().isInterrupted()) {
+      List<ReleaseMessage> messages = releaseMessageRepository.findFirst100ByMessageAndIdLessThanOrderByIdAsc(
+          releaseMessage.getMessage(), releaseMessage.getId());
+
+      releaseMessageRepository.delete(messages);
+      hasMore = messages.size() == 100;
+
+      messages.forEach(toRemove -> Tracer.logEvent(
+          String.format("ReleaseMessage.Clean.%s", toRemove.getMessage()), String.valueOf(toRemove.getId())));
+    }
+  }
+
+  void stopClean() {
+    cleanStopped.set(true);
+  }
 }

apollo-biz/src/main/java/com/ctrip/framework/apollo/biz/repository/ReleaseMessageRepository.java

@@ -19,6 +19,8 @@ public interface ReleaseMessageRepository extends PagingAndSortingRepository<Rel
 
   ReleaseMessage findTopByMessageInOrderByIdDesc(Collection<String> messages);
 
+  List<ReleaseMessage> findFirst100ByMessageAndIdLessThanOrderByIdAsc(String message, Long id);
+
   @Query("select message, max(id) as id from ReleaseMessage where message in :messages group by message")
   List<Object[]> findLatestReleaseMessagesGroupByMessages(@Param("messages") Collection<String> messages);
 }