我尝试使用 .Net 客户端在事件存储集群上写入和读取事件。因此,我设置了 3 个本地 docker 容器来测试对集群的写入和读取。我已经设置了集群,它说它还活着(1 个主节点和 2 个从节点)。
不幸的是,当我尝试从 eventstore 上的 .Net 客户端写入或读取时,连接一直在打开和关闭,并且没有调用到达商店。如何连接到集群以写入和读取事件?
如果我使用没有集群的事件存储,则方法可以正常工作,我可以创建和使用事件。使用集群,我只收到以下日志消息:
Start connecting to Eventstore
Start reading events:
[04,06:49:19.771,INFO] Discovering: found best choice [172.16.1.103:1112,n/a] (Master).
[04,06:49:19.771,INFO] Discovering attempt 1/10 successful: best candidate is [172.16.1.103:1112, n/a].
[10,06:49:20.396,INFO] ClientAPI TcpConnection closed [06:49:20.396: N172.16.1.103:1112, L, {220a086e-3394-49c3-affa-b3d6fc385ea2}]:
[10,06:49:20.396,INFO] Received bytes: 0, Sent bytes: 0
[10,06:49:20.396,INFO] Send calls: 0, callbacks: 0
[10,06:49:20.396,INFO] Receive calls: 0, callbacks: 0
[10,06:49:20.396,INFO] Close reason: [Success] Connection establishment timeout.
[10,06:49:20.397,DEBUG] TcpPackageConnection: connection [172.16.1.103:1112, L, {220a086e-3394-49c3-affa-b3d6fc385ea2}] was closed cleanly.
[10,06:49:22.564,INFO] Discovering attempt 1/10 failed: no candidate found.
[10,06:49:23.068,INFO] Discovering: found best choice [172.16.1.103:1112,n/a] (Master).
[10,06:49:23.068,INFO] Discovering attempt 2/10 successful: best candidate is [172.16.1.103:1112, n/a].
如果我使用 WebUI 将事件写入集群事件存储,它将接受它,然后可以在事件存储的 WebUI 中找到该事件。
如果您需要详细信息,以下是我的设置:
eventstore 集群的设置如eventstore docs中所述。通过设置集群设置为不使用 dns discover-via-dns=false
。这是我.yaml
的 docker 设置文件:
version: '3'
services:
eventstore1:
image: "eventstore/eventstore:latest"
environment:
EVENTSTORE_DISABLE_HTTP_CACHING: "True"
EVENTSTORE_RUN_PROJECTIONS: ALL
EVENTSTORE_CLUSTER_SIZE: 3
EVENTSTORE_INT_IP: 172.16.1.101
EVENTSTORE_EXT_IP: 172.16.1.101
EVENTSTORE_INT_TCP_PORT: 1111
EVENTSTORE_EXT_TCP_PORT: 1112
EVENTSTORE_INT_HTTP_PORT: 2113
EVENTSTORE_EXT_HTTP_PORT: 2114
EVENTSTORE_DISCOVER_VIA_DNS: "False"
EVENTSTORE_GOSSIP_SEED: 172.16.1.102:2113,172.16.1.103:2113
EVENTSTORE_INT_HTTP_PREFIXES: "http://*:2113/"
EVENTSTORE_EXT_HTTP_PREFIXES: "http://*:2114/"
ports:
- 1111:1111
- 1112:1112
- 2113:2113
- 2114:2114
networks:
app_net:
ipv4_address: 172.16.1.101
eventstore2:
image: "eventstore/eventstore:latest"
environment:
EVENTSTORE_DISABLE_HTTP_CACHING: "True"
EVENTSTORE_RUN_PROJECTIONS: ALL
EVENTSTORE_CLUSTER_SIZE: 3
EVENTSTORE_INT_IP: 172.16.1.102
EVENTSTORE_EXT_IP: 172.16.1.102
EVENTSTORE_INT_TCP_PORT: 1111
EVENTSTORE_EXT_TCP_PORT: 1112
EVENTSTORE_INT_HTTP_PORT: 2113
EVENTSTORE_EXT_HTTP_PORT: 2114
EVENTSTORE_DISCOVER_VIA_DNS: "False"
EVENTSTORE_GOSSIP_SEED: 172.16.1.101:2113,172.16.1.103:2113
EVENTSTORE_INT_HTTP_PREFIXES: "http://*:2113/"
EVENTSTORE_EXT_HTTP_PREFIXES: "http://*:4114/"
ports:
- 3111:1111
- 3112:1112
- 4113:2113
- 4114:2114
networks:
app_net:
ipv4_address: 172.16.1.102
eventstore3:
image: "eventstore/eventstore:latest"
environment:
EVENTSTORE_DISABLE_HTTP_CACHING: "True"
EVENTSTORE_RUN_PROJECTIONS: ALL
EVENTSTORE_CLUSTER_SIZE: 3
EVENTSTORE_INT_IP: 172.16.1.103
EVENTSTORE_EXT_IP: 172.16.1.103
EVENTSTORE_INT_TCP_PORT: 1111
EVENTSTORE_EXT_TCP_PORT: 1112
EVENTSTORE_INT_HTTP_PORT: 2113
EVENTSTORE_EXT_HTTP_PORT: 2114
EVENTSTORE_DISCOVER_VIA_DNS: "False"
EVENTSTORE_GOSSIP_SEED: 172.16.1.101:2113,172.16.1.102:2113
EVENTSTORE_INT_HTTP_PREFIXES: "http://*:2113/"
EVENTSTORE_EXT_HTTP_PREFIXES: "http://*:2114/"
ports:
- 5111:1111
- 5112:1112
- 6113:2113
- 6114:2114
networks:
app_net:
ipv4_address: 172.16.1.103
networks:
app_net:
external: true
为了测试连接以及是否可以在 eventstore 集群上读写,我创建了以下小应用程序:
主程序:
class Program
{
const string STREAM = "MyTestStream";
private static ConnectionFactory _connectionFactory = new ConnectionFactory();
static void Main(string[] args)
{
var useCluster = true;
Console.WriteLine("Start connecting to Eventstore");
var writeEvents = new WriteEvents(_connectionFactory);
writeEvents.Write(useCluster, STREAM).Wait();
Console.WriteLine("Start reading events:");
var readEvents = new ReadEvents(_connectionFactory);
readEvents.Read(useCluster, STREAM).Wait();
Console.WriteLine("Check if events are written and press key to exit");
Console.ReadKey();
}
}
连接工厂:
public class ConnectionFactory
{
public IEventStoreConnection SingleConnection()
{
return EventStoreConnection.Create(ConnectionSettings.Create(),
new IPEndPoint(IPAddress.Parse("127.0.0.1"), 1113));
}
public IEventStoreConnection ClusterConnection()
{
return EventStoreConnection.Create(
ConnectionSettings.Create().KeepReconnecting().UseConsoleLogger()
.SetGossipSeedEndPoints(
new IPEndPoint(IPAddress.Parse("127.0.0.1"), 2113),
new IPEndPoint(IPAddress.Parse("127.0.0.1"), 4113),
new IPEndPoint(IPAddress.Parse("127.0.0.1"), 6113))
.WithConnectionTimeoutOf(TimeSpan.FromMilliseconds(500))
);
}
public IEventStoreConnection CreateConnection(bool useCluster)
{
if (useCluster)
{
return ClusterConnection();
}
return SingleConnection();
}
}
集群的连接是按照eventstore 文档中的描述创建的,但是该文档与我使用的最新 nuget 版本并非 100% 一致。
nuget 版本:
<PackageReference Include="EventStore.ClientAPI.NetCore" Version="4.1.0.23" />
写事件:
public class WriteEvents
{
private ConnectionFactory _connectionFactory;
public WriteEvents(ConnectionFactory connectionFactory)
{
_connectionFactory = connectionFactory;
}
public async Task Write(bool useCluster, string streamName)
{
using (var conn = _connectionFactory.CreateConnection(useCluster))
{
await conn.ConnectAsync();
for (var x = 0; x < 100; x++)
{
await conn.AppendToStreamAsync(streamName,
ExpectedVersion.Any,
GetEventDataFor(x));
Console.WriteLine("event " + x + " written.");
}
}
}
private IEnumerable<EventData> GetEventDataFor(int i)
{
yield return new EventData(
Guid.NewGuid(),
"MyTestEvent",
true,
Encoding.ASCII.GetBytes("{'somedata' : " + i + "}"),
Encoding.ASCII.GetBytes("{'metadata' : " + i + "}")
);
}
}
阅读事件:
public class ReadEvents
{
private ConnectionFactory _connectionFactory;
public ReadEvents(ConnectionFactory connectionFactory)
{
_connectionFactory = connectionFactory;
}
public async Task Read(bool useCluster, string streamName)
{
using (var conn = _connectionFactory.CreateConnection(useCluster))
{
await conn.ConnectAsync();
var slice = await conn.ReadStreamEventsForwardAsync(streamName, 0, 100, false);
foreach (var evt in slice.Events)
{
Console.WriteLine($"Received event. Type: '{evt.Event.EventType}', Data: '{Encoding.UTF8.GetString(evt.Event.Data)}'");
}
}
}
}