我尝试使用.Net客户端在事件存储群集上写入和读取事件。
因此，我设置了3个本地docker容器来测试对集群的写入和读取。我已经设置了集群，它说它还活着(1个主节点和2个从节点)。
不幸的是，当我尝试在事件存储区上从.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 docs中所述设置事件存储集群。
通过设置
discover-via-dns=false将集群设置为而不是使用dns。这是我的docker设置的.yaml文件: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
为了测试连接以及是否可以在事件存储群集上进行读写，我创建了以下小型应用程序:
主程序:
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();
}
}
ConnectionFactory:
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 documentation所述创建，但是文档未与我使用的最新nuget版本100％对齐。
nuget版本:
<PackageReference Include="EventStore.ClientAPI.NetCore" Version="4.1.0.23" />
WriteEvents:
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 + "}")
);
}
}
ReadEvents:
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)}'");
}
}
}
}
最佳答案
这与客户端类型无关。如果使用非NetCore，也会得到相同的行为。
当客户端尝试连接到群集中的一个节点时，EventStore会回复一个无法访问的IP。
在某些情况下，例如当您使用软件编排(如Kubernetes或群集)时，您的客户端需要托管在同一覆盖网络中，才能使用Tcp客户端连接到群集。
在开发应用程序/微服务时，您可以使用连接字符串连接到不在Kubernetes，Swarm或本地Docker引擎中托管的单节点EventStore。在Kubernetes，swarm或Docker-Compose环境中部署应用程序时，您可以使用八卦种子设置应用程序的连接字符串，并使用集群主机的ips。
使用集群连接的C#客户端
namespace TestClusterConnection
{
class Program
{
private const string Stream = "MyTestStream";
static void Main(string[] args)
{
try
{
var useCluster = true;
Console.WriteLine("Start connecting to Eventstore");
Write(useCluster, Stream).Wait();
Console.WriteLine("Start reading events:");
Read(useCluster, Stream).Wait();
Console.WriteLine("Check if events are written and press key to exit");
}
catch (Exception e)
{
Console.WriteLine(e.GetBaseException().Message);
}
Console.ReadKey();
}
public static async Task Read(bool useCluster, string streamName)
{
using (var conn = 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)}'");
}
}
private static async Task Write(bool useCluster, string streamName)
{
using (var conn = 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 static IEnumerable<EventData> GetEventDataFor(int i)
{
yield return new EventData(
Guid.NewGuid(),
"MyTestEvent",
true,
Encoding.ASCII.GetBytes("{'somedata' : " + i + "}"),
Encoding.ASCII.GetBytes("{'metadata' : " + i + "}")
);
}
private static IEventStoreConnection SingleConnection()
{
return EventStoreConnection.Create(ConnectionSettings.Create(),
new IPEndPoint(IPAddress.Parse("127.0.0.1"), 1113));
}
private static IEventStoreConnection ClusterConnection()
{
return EventStoreConnection.Create(
ConnectionSettings.Create().KeepRetrying().KeepReconnecting().UseConsoleLogger()
.SetGossipSeedEndPoints(
new IPEndPoint(IPAddress.Parse("172.19.0.2"), 2112),
new IPEndPoint(IPAddress.Parse("172.19.0.3"), 2112),
new IPEndPoint(IPAddress.Parse("172.19.0.4"), 2112))
.SetHeartbeatInterval(TimeSpan.FromSeconds(3))
.SetHeartbeatTimeout(TimeSpan.FromSeconds(6))
.WithConnectionTimeoutOf(TimeSpan.FromSeconds(10))
);
}
private static IEventStoreConnection CreateConnection(bool useCluster)
{
return useCluster ? ClusterConnection() : SingleConnection();
}
}
}
Docker文件
FROM mono:4.6.2.16
ADD . /home/TestClusterConnection
CMD [ "mono", "home/TestClusterConnection/TestClusterConnection.exe" ]
docker-compose.yaml
version: '3.4'
services:
esclienttest:
image: testclient
build:
context: .
dockerfile: Dockerfile
depends_on:
- eventstore1
- eventstore2
- eventstore3
eventstore1:
image: eventstore/eventstore:release-4.1.0
hostname: eventstore1
ports:
- 1113:1113
- 2112:2112
environment:
EVENTSTORE_CLUSTER_DNS: eventstore1
EVENTSTORE_CLUSTER_SIZE: 3
EVENTSTORE_CLUSTER_GOSSIP_PORT: 2112
EVENTSTORE_EXT_IP_ADVERTISE_AS: 172.19.0.2
eventstore2:
image: eventstore/eventstore:release-4.1.0
hostname: eventstore2
environment:
EVENTSTORE_CLUSTER_DNS: eventstore1
EVENTSTORE_CLUSTER_SIZE: 3
EVENTSTORE_CLUSTER_GOSSIP_PORT: 2112
EVENTSTORE_EXT_IP_ADVERTISE_AS: 172.19.0.3
eventstore3:
image: eventstore/eventstore:release-4.1.0
hostname: eventstore3
environment:
EVENTSTORE_CLUSTER_DNS: eventstore1
EVENTSTORE_CLUSTER_SIZE: 3
EVENTSTORE_CLUSTER_GOSSIP_PORT: 2112
EVENTSTORE_EXT_IP_ADVERTISE_AS: 172.19.0.4
仅用于测试，可能需要进一步的网络设置。
希望这可以帮助。
里卡多
关于.net - 如何通过.Net客户端在事件存储群集上写入和读取事件，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/51375166/