前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >聊聊artemis的persistenceEnabled

聊聊artemis的persistenceEnabled

作者头像
code4it
发布2020-02-24 09:38:54
3470
发布2020-02-24 09:38:54
举报
文章被收录于专栏:码匠的流水账码匠的流水账

本文主要研究一下artemis的persistenceEnabled

persistenceEnabled

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImpl.java

代码语言:javascript
复制
public class ConfigurationImpl implements Configuration, Serializable {

   //......

   private boolean persistenceEnabled = ActiveMQDefaultConfiguration.isDefaultPersistenceEnabled();

   public boolean isPersistenceEnabled() {
      return persistenceEnabled;
   }

   public ConfigurationImpl setPersistenceEnabled(final boolean enable) {
      persistenceEnabled = enable;
      return this;
   }

   //......
}  
  • ConfigurationImpl定义了persistenceEnabled属性,默认为ActiveMQDefaultConfiguration.isDefaultPersistenceEnabled()(true)

createStorageManager

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java

代码语言:javascript
复制
public class ActiveMQServerImpl implements ActiveMQServer {

   //......

   protected StorageManager createStorageManager() {
      if (configuration.isPersistenceEnabled()) {
         if (configuration.getStoreConfiguration() != null && configuration.getStoreConfiguration().getStoreType() == StoreConfiguration.StoreType.DATABASE) {
            JDBCJournalStorageManager journal = new JDBCJournalStorageManager(configuration, getCriticalAnalyzer(), getScheduledPool(), executorFactory, ioExecutorFactory, shutdownOnCriticalIO);
            this.getCriticalAnalyzer().add(journal);
            return journal;
         } else {
            // Default to File Based Storage Manager, (Legacy default configuration).
            JournalStorageManager journal = new JournalStorageManager(configuration, getCriticalAnalyzer(), executorFactory, scheduledPool, ioExecutorFactory, shutdownOnCriticalIO);
            this.getCriticalAnalyzer().add(journal);
            return journal;
         }
      }
      return new NullStorageManager();
   }

   //......
}
  • ActiveMQServerImpl的createStorageManager方法在configuration.isPersistenceEnabled()为true时创建的StorageManager是JDBCJournalStorageManager或者JournalStorageManager;否则创建的StorageManager是NullStorageManager

processRoute

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java

代码语言:javascript
复制
public class PostOfficeImpl implements PostOffice, NotificationListener, BindingsFactory {

   //......

   public void processRoute(final Message message,
                            final RoutingContext context,
                            final boolean direct) throws Exception {
      final List<MessageReference> refs = new ArrayList<>();

      Transaction tx = context.getTransaction();

      Long deliveryTime = message.getScheduledDeliveryTime();

      for (Map.Entry<SimpleString, RouteContextList> entry : context.getContexListing().entrySet()) {
         PagingStore store = pagingManager.getPageStore(entry.getKey());

         if (store != null && storageManager.addToPage(store, message, context.getTransaction(), entry.getValue())) {
            if (message.isLargeMessage()) {
               confirmLargeMessageSend(tx, message);
            }

            // We need to kick delivery so the Queues may check for the cursors case they are empty
            schedulePageDelivery(tx, entry);
            continue;
         }

         for (Queue queue : entry.getValue().getNonDurableQueues()) {
            MessageReference reference = MessageReference.Factory.createReference(message, queue);

            if (deliveryTime != null) {
               reference.setScheduledDeliveryTime(deliveryTime);
            }
            refs.add(reference);

            message.incrementRefCount();
         }

         Iterator<Queue> iter = entry.getValue().getDurableQueues().iterator();

         while (iter.hasNext()) {
            Queue queue = iter.next();

            MessageReference reference = MessageReference.Factory.createReference(message, queue);

            if (context.isAlreadyAcked(context.getAddress(message), queue)) {
               reference.setAlreadyAcked();
               if (tx != null) {
                  queue.acknowledge(tx, reference);
               }
            }

            if (deliveryTime != null) {
               reference.setScheduledDeliveryTime(deliveryTime);
            }
            refs.add(reference);

            if (message.isDurable()) {
               int durableRefCount = message.incrementDurableRefCount();

               if (durableRefCount == 1) {
                  if (tx != null) {
                     storageManager.storeMessageTransactional(tx.getID(), message);
                  } else {
                     storageManager.storeMessage(message);
                  }

                  if (message.isLargeMessage()) {
                     confirmLargeMessageSend(tx, message);
                  }
               }

               if (tx != null) {
                  storageManager.storeReferenceTransactional(tx.getID(), queue.getID(), message.getMessageID());

                  tx.setContainsPersistent();
               } else {
                  storageManager.storeReference(queue.getID(), message.getMessageID(), !iter.hasNext());
               }

               if (deliveryTime != null && deliveryTime > 0) {
                  if (tx != null) {
                     storageManager.updateScheduledDeliveryTimeTransactional(tx.getID(), reference);
                  } else {
                     storageManager.updateScheduledDeliveryTime(reference);
                  }
               }
            }

            message.incrementRefCount();
         }
      }

      if (tx != null) {
         tx.addOperation(new AddOperation(refs));
      } else {
         // This will use the same thread if there are no pending operations
         // avoiding a context switch on this case
         storageManager.afterCompleteOperations(new IOCallback() {
            @Override
            public void onError(final int errorCode, final String errorMessage) {
               ActiveMQServerLogger.LOGGER.ioErrorAddingReferences(errorCode, errorMessage);
            }

            @Override
            public void done() {
               context.processReferences(refs, direct);
            }
         });
      }
   }

   //......
}
  • PostOfficeImpl的processRoute方法会判断message.isDurable(),若为true且durableRefCount为1则会执行storageManager.storeMessage或者storageManager.storeMessageTransactional方法

storeMessage

AbstractJournalStorageManager

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java

代码语言:javascript
复制
public abstract class AbstractJournalStorageManager extends CriticalComponentImpl implements StorageManager {

   //......

   protected Journal messageJournal;

   //......

   public void storeMessage(final Message message) throws Exception {
      if (message.getMessageID() <= 0) {
         // Sanity check only... this shouldn't happen unless there is a bug
         throw ActiveMQMessageBundle.BUNDLE.messageIdNotAssigned();
      }

      readLock();
      try {
         // Note that we don't sync, the add reference that comes immediately after will sync if
         // appropriate

         if (message.isLargeMessage()) {
            messageJournal.appendAddRecord(message.getMessageID(), JournalRecordIds.ADD_LARGE_MESSAGE, LargeMessagePersister.getInstance(), message, false, getContext(false));
         } else {
            messageJournal.appendAddRecord(message.getMessageID(), JournalRecordIds.ADD_MESSAGE_PROTOCOL, message.getPersister(), message, false, getContext(false));
         }
      } finally {
         readUnLock();
      }
   }

   //......
}
  • JDBCJournalStorageManager及JournalStorageManager都继承了AbstractJournalStorageManager,其storeMessage方法会调用messageJournal.appendAddRecord方法;二者的messageJournal实现不同,一个是JDBCJournalImpl,一个是JournalImpl

NullStorageManager

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java

代码语言:javascript
复制
public class NullStorageManager implements StorageManager {

   //......

   public void storeMessage(final Message message) throws Exception {
   }

   //......
}
  • NullStorageManager实现了StorageManager接口,其storeMessage为空方法

小结

ConfigurationImpl定义了persistenceEnabled属性,默认为ActiveMQDefaultConfiguration.isDefaultPersistenceEnabled()(true);ActiveMQServerImpl的createStorageManager方法在configuration.isPersistenceEnabled()为true时创建的StorageManager是JDBCJournalStorageManager或者JournalStorageManager;否则创建的StorageManager是NullStorageManager;PostOfficeImpl的processRoute方法会判断message.isDurable(),若为true且durableRefCount为1则会执行storageManager.storeMessage或者storageManager.storeMessageTransactional方法

doc

  • ConfigurationImpl
本文参与 腾讯云自媒体分享计划,分享自微信公众号。
原始发表:2020-01-26,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 码匠的流水账 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • persistenceEnabled
  • createStorageManager
  • processRoute
  • storeMessage
    • AbstractJournalStorageManager
      • NullStorageManager
      • 小结
      • doc
      领券
      问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档