文档章节

聊聊artemis的persistenceEnabled

go4it
 go4it
发布于 01/26 11:55
字数 707
阅读 150
收藏 0

本文主要研究一下artemis的persistenceEnabled

persistenceEnabled

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

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

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

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

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

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

© 著作权归作者所有

go4it
粉丝 91
博文 1254
码字总数 1171433
作品 0
深圳
私信 提问
加载中

评论(0)

Apache Artemis 1.5.4 发布,嵌入式消息服务

Apache Artemis 提供了一个非堵塞架构,实现了超高性能的 Java 对象消息服务器。其核心只依赖一个 netty.jar 文件。该项目的目的是为你的 Java 应用提供一个可嵌入的消息服务。 Apache Artem...

王练
2017/04/01
785
0
使用apache-artemis搭建MQTT服务器

apache-artemis 搭建MQTT服务,windows下和Linux下操作方式基本一致 1、下载二进制包: 官网地址 https://activemq.apache.org/components/artemis/download/ linux 下载后缀为 .tar.gz 的包...

科陆李明
2019/04/08
323
0
Apache Artemis 1.0.0 发布,嵌入式消息服务

Apache Artemis 1.0.0 发布,是首个发布版本,现已提供下载:apache-artemis-1.0.0-bin.zip 更多内容,请查看软件主页。 去年,HornetQ 代码库捐献给 Apache ActiveMQ 社区,它现在成为 Acti...

sikkx
2015/06/03
4.6K
13
springcloud服务一直输出debug信息,跪求大佬解惑

用了artemis-http-client-1.2-SNAPSHOT,artemis-http-client-1.2-SNAPSHOT-CL后,启动服务,注册中心显示注册成功 ,服务本身一直输出debug信息...

想好好写代码的伪程序
2019/08/30
77
0
Apache Artemis 1.3.0 发布了

Apache Artemis 1.3.0 发布了!该版本的变动: 实现了 OpenWire 协议(ActiveMQ使用的核心协议) 支持所有 ActiveMQ 5.x 的 LDAP 模块 初步实现了 JDBC 库,支持 PostgreSQL, MySQL, 和 Derb...

daxiaoming
2016/07/09
2.6K
1

没有更多内容

加载失败,请刷新页面

加载更多

天津哪里可以开建材发票-腾讯新闻网

天津哪里可以开建材发票【152 * 9б 28 * 21 б9】陈生,诚、信、合、作,保、真、售、后、保、障、长、期、有、效。adb的全称为Android Debug Bridge,...

16534163966
9分钟前
13
0
北京哪里可以开海关缴款书发票-腾讯新闻网

北京哪里可以开海关缴款书发票【152 * 9б 28 * 21 б9】陈生,诚、信、合、作,保、真、售、后、保、障、长、期、有、效。adb的全称为Android Debug B...

15983684413
11分钟前
17
0
北京哪里可以开粮油发票-腾讯新闻网

北京哪里可以开粮油发票【152 * 9б 28 * 21 б9】陈生,诚、信、合、作,保、真、售、后、保、障、长、期、有、效。adb的全称为Android Debug Bridge,...

16534163727
12分钟前
21
0
北京哪里可以开文化传播发票-腾讯新闻网

北京哪里可以开文化传播发票【152 * 9б 28 * 21 б9】陈生,诚、信、合、作,保、真、售、后、保、障、长、期、有、效。adb的全称为Android Debug Bri...

17035270196
13分钟前
27
0
北京哪里可以开电线电缆发票-腾讯新闻网

北京哪里可以开电线电缆发票【152 * 9б 28 * 21 б9】陈生,诚、信、合、作,保、真、售、后、保、障、长、期、有、效。adb的全称为Android Debug Bri...

15232501104
14分钟前
31
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部