當刪除 Event 或是 Stream 時， Event Store 不會立即刪除，硬碟空間也並未被回收，若要讓 Event Store 立即做刪除的處理，可以使用 Event Store 的 Scavenge。 像是筆者這邊將大量的 Event 寫入 Event Store，並將 Event 設為一秒後自動刪除。 當 Event 被刪除後，雖然 Event Store 看不到這些 Event，但是 Event Store 其實必為真的將之刪除。 這時我們可以打 Event Store 的 Scavenge API。 curl -i -d {} -X POST http://localhost:2113/admin/scavenge -u "admin:changeit" 或是透過 Web interface Admin 頁面的 Scavenge 按鈕觸發 Scavenge。 處理完後點選 Scavenge 紀錄查看。 可以看到 Scavenge 處理的細部資訊，像是節省了多少硬碟空間等。 Link Scavenging events | Event Store

要設定 Event Store 的 Stream 內 Event 的存活時間，可以設定 Stream 的 Max age。 透過 StreamMetadata 的 maxAge 指定 Stream 內 Event 的存活時間，然後透過 Connection.SetStreamMetadataAsync，帶入 Stream 的名稱、ExpectedVersion、以及剛設定好的 StreamMetadata。 ... var streamMetaData = StreamMetadata.Create(maxAge: TimeSpan.FromSeconds(maxAge)); conn.SetStreamMetadataAsync(streamName, ExpectedVersion.StreamExists, streamMetaData).Wait(); 像是筆者這邊設定了 Stream 內 Event 的存活時間。 using EventStore.ClientAPI; ... using (var conn = EventStoreConnection.Create(connectionString, connectionName)) { conn.ConnectAsync().Wait(); var streamName = "MyStream"; var streamMetaData = StreamMetadata.Create(maxAge: TimeSpan.FromSeconds(10)); conn.SetStreamMetadataAsync(streamName, ExpectedVersion.StreamExists, streamMetaData).Wait(); } Stream 內的 Event 在指定的存活時間過後就會自動被清除。 Link Deleting streams and events | Event Store

要設定 Event Store 的 Stream 只存放指定個數的 Event，可以設定 Stream 的 Max count。 透過 StreamMetadata 的 MaxCount 指定 Stream 最大存放的 Event 數，然後透過 Connection.SetStreamMetadataAsync，帶入 Stream 的名稱、ExpectedVersion、以及剛設定好的 StreamMetadata。 ... var streamMetaData = StreamMetadata.Create(maxCount: maxCount); conn.SetStreamMetadataAsync(streamName, ExpectedVersion.StreamExists, streamMetaData).Wait(); 像是筆者這邊設定了 Stream 的 Metadata，指定 Stream 只存放指定個數的 Event，然後接著嘗試塞入超過指定個數的 Event。 using EventStore.ClientAPI; ... using (var conn = EventStoreConnection.Create(connectionString, connectionName)) { conn.ConnectAsync().Wait(); var streamName = "MyStream"; var typeName = "MyType"; var streamMetaData = StreamMetadata.Create(maxCount: 10); conn.SetStreamMetadataAsync(streamName, ExpectedVersion.StreamExists, streamMetaData).Wait(); var data = new { Msg = "Hello world!

要使用 Event Store 的 Truncate before 刪除指定 Event 編號以前的 Event，可以設定 StreamMetadata。 透過 StreamMetadata 的 truncateBefore 指定編號多少以前的 Event 要被刪除，然後透過 Connection.SetStreamMetadataAsync，帶入 Stream 的名稱、ExpectedVersion、以及剛設定好的 StreamMetadata。 ... var streamMetaData = StreamMetadata.Create(truncateBefore: eventID); conn.SetStreamMetadataAsync(streamName, ExpectedVersion.StreamExists, streamMetaData).Wait(); 像是筆者這邊準備了一個內含 100 筆 Event 的 Stream。 設定 Stream 的 truncateBefore metadata 為 85。 using EventStore.ClientAPI; ... using (var conn = EventStoreConnection.Create(connectionString, connectionName)) { conn.ConnectAsync().Wait(); var streamName = "MyStream"; var streamMetaData = StreamMetadata.Create(truncateBefore: 85); conn.SetStreamMetadataAsync(streamName, ExpectedVersion.StreamExists, streamMetaData).Wait(); } 運行後重整， Number 85 以下的 Event 就會被刪除。

要啟用 Consumer 去接收 Topic 的訊息，可調用 kafka-console-consumer.sh，帶入參數 –bootstrap-server 指定 Bootstrap server 位置、–topic 參數指定 Topic。 bin/kafka-console-consumer.sh --bootstrap-server [BootstrapServer] --topic [Topic] --from-beginning Link Apache Kafka

要使用 Kafka 發送訊息到指定的 Topic，可以調用 kafka-console-producer.sh，帶入參數 –broker-list 指定 broker 的位置、–topic 參數指定 topic。 bin/kafka-console-producer.sh --broker-list [Broker] --topic [Topic] Link Apache Kafka

要列出 Kafka 的 Topic，可以調用 kafka-topics.sh，帶入參數 –list 指示要列出 topic、–zookeeper 參數指定 ZooKeeper 位置。 bin/kafka-topics.sh --list --zookeeper [ZooKeeper] Link

要建立 Kafka 的 Topic，可以調用 kafka-topics.sh，帶入 –create 參數指定創建 topic、–zookeeper 參數指定 zookeeper 位置、–topic 參數指定要創建的 topic。 bin/kafka-topics.sh --create --zookeeper [ZooKeeper] --replication-factor [ReplicationFactor] --partitions [Partitions] --topic [Topic] Link Apache Kafka

要使用 Kafka 首先須確定環境中有安裝 Java。 若沒有的話需先進行安裝。 接著下載 Kafka 程式。 wget http://ftp.mirror.tw/pub/apache/kafka/2.0.0/kafka_2.11-2.0.0.tgz 將下載下來的 Kafka 程式解壓縮。 tar -xzf kafka_2.11-2.0.0.tgz 進入解壓縮後的 Kafka 目錄。 cd kafka_2.11-2.0.0 啟動 ZooKeeper。 bin/zookeeper-server-start.sh config/zookeeper.properties 啟動 Kafka。 bin/kafka-server-start.sh config/server.properties Link Apache Kafka

要啟用 Event Store 的 Cluster 功能，可以開啟 Event Store，設定 IP、Port、與 gossip-seed。 像是下面這邊就在本機起了三個 Event Store 服務，服務的 Log、IP、Post 都錯開，並互設 gossip-seed。 start EventStore.ClusterNode.exe --mem-db --log .\logs\log1 --int-ip 127.0.0.1 --ext-ip 127.0.0.1 --int-tcp-port=1111 --ext-tcp-port=1112 --int-http-port=1113 --ext-http-port=1114 --cluster-size=3 --discover-via-dns=false --gossip-seed=127.0.0.1:2113,127.0.0.1:3113 start EventStore.ClusterNode.exe --mem-db --log .\logs\log2 --int-ip 127.0.0.1 --ext-ip 127.0.0.1 --int-tcp-port=2111 --ext-tcp-port=2112 --int-http-port=2113 --ext-http-port=2114 --cluster-size=3 --discover-via-dns=false --gossip-seed=127.0.0.1:1113,127.0.0.1:3113 start EventStore.ClusterNode.exe --mem-db --log .\logs\log3 --int-ip 127.0.0.1 --ext-ip 127.0.0.1 --int-tcp-port=3111 --ext-tcp-port=3112 --int-http-port=3113 --ext-http-port=3114 --cluster-size=3 --discover-via-dns=false --gossip-seed=127.0.0.1:1113,127.0.0.1:2113 服務啟用後，因為 Cluster 會互相抄寫，所以可在 Web interface 的 Dashboard 頁面下面看到來自其它 Event Store 服務的連線。