使用NATS及其nats.c客户端简单示例用法（nats.c的API接口）

前言

什么是NATS

首先我们先来了解一下什么是NATS,NATS的话是一个开源、轻量级、高性能的分布式消息中间件，主要是基于go语言开发的，是一种简单，安全且高性能的通信系统，适用于数字系统，服务和设备。NATS是云原生计算基金会（CNCF）的一部分。NATS有40多个客户端语言实现，其服务器可以在本地，云中，边缘甚至Raspberry Pi上运行。NATS可以保护和简化现代分布式系统的设计和操作。它支持的消息传递模型有发布订阅、请求回复、队列订阅等，功能则有发布订阅模型、服务器集群、自动订阅者、基于文本协议等。

NATS 客户端应用程序

开发人员在其应用程序代码中使用其中一个NATS客户端库（NATS客户端有很多语言编写，可以根据需求选其中一种来进行操作），以允许他们在应用程序实例之间或完全独立的应用程序之间发布、订阅、请求和回复。这些应用程序在本手册中通常称为“客户端应用程序”，有时也称为"客户端”(因为从 NATS服务器的角度来看,它们是客户端)。

NATS 服务基础架构

NATS服务由一个或多个NATS服务器进程提供，这些进程配置为相互互连并提供NATS服务基础结构。NATS服务基础设施可以从在终端设备上运行的单个NATS服务器进程（nats服务器进程的大小小于20MB! )一直扩展到一个由许多集群组成的公共全球超级集群，这些集群跨越所有主要云提供商和世界所有地区，如Synadia的NGS。

将NATS客户端应用程序连接到NATS服务器

要将NATS客户端应用程序与NATS服务连接，然后订阅或发布消息到主题，只需为其配置:

1.网址: A"NATS URL".这是一个字符串(采用URL格式)，用于指定可以访问NATS服务器的IP地址和端口，以及要建立的连接类型(纯 TCP、TLS或 Websocket)。

2.身份验证(如果需要):认证应用程序的详细信息，以通过NATS服务器标识自身。NATS支持多种身份验证方案（用户名/密码，分散式JWT，令牌，TLS证书和带质询的Nkey)。

简单的消息传递设计

NATS使应用程序能够通过发送和接收消息来轻松进行通信。这些邮件由主题字符串寻址和标识，并且不依赖于网络位置。

数据被编码并框定为消息，并由发布者发送。消息由一个或多个订阅者接收、解码和处理。

通过这种简单的设计，NATS允许程序共享常见的消息处理代码，隔离资源和相互依赖关系，并通过轻松处理消息量的增加(无论是服务请求还是流数据)进行扩展。

以上来源于nats官方文档，有兴趣的可以去官方查看详情NATS官方文档

NATS客户端

这里主要是基于在Linux中使用nats.c客户端的用法，如想查看其他语言的客户端用法，可通过该https://github.com/nats-io进行查看。

安装

首先我们在Linux中下载该源码，使用以下命令

sudo git clone git@github.com:nats-io/nats.c.git .
1

或者使用前往https://github.com/nats-io/nats.c进行下载

下载好了之后，我们可以直接进入到nats.c目录里，使用命令cd nats.c

然后再使用命令mkdir build 创建build目录,再加入到该build目录里，然后执行命令cmake …

开始

在该目录里，该目录有一组功能齐全的简单示例，但非常简单。目标是演示API的易用性。加入到nats.c目录里之后，再加入到该目录里，使用命令 cd examples/getstarted，然后就可以看到里面的文件，都是一些简单的示例文件。

目录中有一组更复杂的示例，也可用于对客户端库进行基准测试。在 examples/ 目录里

基本用法

natsConnection      *nc  = NULL;	//nats连接
natsSubscription    *sub = NULL;	//nats订阅
natsMsg             *msg = NULL;

// 连接到本地运行的默认nats Server
natsConnection_ConnectTo(&nc, NATS_DEFAULT_URL);

// 使用用户名和密码连接到服务器
natsConnection_ConnectTo(&nc, "nats://ivan:secret@localhost:4222");

// 使用令牌身份验证连接到服务器
natsConnection_ConnectTo(&nc, "nats://myTopSecretAuthenticationToken@localhost:4222");

// 简单的发布者，将给定的字符串发送到主题“foo”
natsConnection_PublishString(nc, "foo", "hello world");

// 发布二进制数据。内容不会被解释为字符串。
char data[] = {1, 2, 0, 4, 5};
natsConnection_Publish(nc, "foo", (const void*) data, 5);

// 主题foo上的简单异步订阅者，调用消息
// handler 'onMsg'时接收消息，并且不提供闭包。
natsConnection_Subscribe(&sub, nc, "foo", onMsg, NULL);

// 简单的同步用户
natsConnection_SubscribeSync(&sub, nc, "foo");

// 使用同步订阅者，获取第一个可用的消息，等待 高达1000毫秒(1秒)
natsSubscription_NextMsg(&msg, sub, 1000);

// 销毁任何接收(异步或同步)或创建的消息
// 通过您的应用程序。注意，如果'msg'为NULL，调用没有效果。
natsMsg_Destroy(msg);

// 取消订阅
natsSubscription_Unsubscribe(sub);

// 销毁订阅(这将释放对象，这可能结果释放内存)。在这个调用之后，对象必须是no
// 再使用
natsSubscription_Destroy(sub);

// 向给定的回复主题发布请求:
natsConnection_PublishRequestString(nc, "foo", "bar", "help!");

// 发送请求(在内部创建收件箱)并自动取消订阅，内部订阅者，
//这意味着订阅者未订阅当收到可能有许多回复者的第一个回复时。
//这个调用将等待1000毫秒(1秒)的应答。
natsConnection_RequestString(&reply, nc, "foo", "help", 1000);

// 关闭连接(但不释放连接对象)
natsConnection_Close(nc);

// 处理完对象后，释放内存。注意，这个调用首先关闭连接，换句话说，
//您可以简单地这个调用取代了natsConnection_Close()后跟destroy调用。
natsConnection_Destroy(nc);

// 消息处理程序
void
onMsg(natsConnection *nc, natsSubscription *sub, natsMsg *msg, void *closure)
{
    // 使用消息getter打印消息:
    printf("Received msg: %s - %.*s\n",
        natsMsg_GetSubject(msg),
        natsMsg_GetDataLength(msg),
        natsMsg_GetData(msg));

    // 别忘了销毁 message!
    natsMsg_Destroy(msg);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

JetStream 基本用法

// 连接到NATS
natsConnection_Connect(&conn, opts);

// 初始化并设置一些JetStream选项
jsOptions jsOpts;
jsOptions_Init(&jsOpts);
jsOpts.PublishAsync.MaxPending = 256;

// 创建 JetStream 上下文
natsConnection_JetStream(&js, conn, &jsOpts);

// 简单流出版商
js_Publish(&pa, js, "ORDERS.scratch", (const void*) "hello", 5, NULL, &jerr);

// 简单的异步流发布程序
for (i=0; i < 500; i++)
{
    js_PublishAsync(js, "ORDERS.scratch", (const void*) "hello", 5, NULL);
}

// 等待最多5秒以接收所有发布确认。
jsPubOptions_Init(&jsPubOpts);
jsPubOpts.MaxWait = 5000;
js_PublishAsyncComplete(js, &jsPubOpts);

// 可以获得所有挂起的发布异步消息的列表，
// 重新发送它们或直接销毁它们。
natsMsgList pending;
s = js_PublishAsyncGetPendingList(&pending, js);
if (s == NATS_OK)
{
    int i;

    for (i=0; i<pending.Count; i++)
    {

        // 可能需要决定是否重新发送这些消息。
        if (your_decision_to_resend(pending.Msgs[i]))
        {

            // 如果呼叫成功，等待。设置Msgs[i]
            // 设置为NULL，这样销毁挂起的列表将不会销毁
            // 库已经收回了所有权
            js_PublishMsgAsync(js, &(pending.Msgs[i]), NULL);
        }
    }

    // 销毁挂起列表对象和仍然在该列表中的所有消息。
    natsMsgList_Destroy(&pending);
}

// 创建异步临时使用者
js_Subscribe(&sub, js, "foo", myMsgHandler, myClosure, &jsOpts, NULL, &jerr);

// 相同，但使用订阅选项要求回调不执行自动ack。
jsSubOptions so;
jsSubOptions_Init(&so);
so.ManualAck = true;
js_Subscribe(&sub, js, "foo", myMsgHandler, myClosure, &jsOpts, &so, &jerr);

// 或者绑定到一个现有的特定流/持久流:
jsSubOptions_Init(&so);
so.Stream = "MY_STREAM";
so.Consumer = "my_durable";
js_Subscribe(&sub, js, "foo", myMsgHandler, myClosure, &jsOpts, &so, &jerr);

// 同步订阅:
js_SubscribeSync(&sub, js, "foo", &jsOpts, &so, &jerr);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

JetStream 基本管理

jsStreamConfig  cfg;	//JetStream 流配置

// 连接到NATS
natsConnection_Connect(&conn, opts);

//创建地方上下文
natsConnection_JetStream(&js, conn, NULL);

//初始化配置结构
jsStreamConfig_Init(&cfg);

// 提供一个名称
cfg.Name = "ORDERS";

// 主题的排列和它的大小
cfg.Subjects = (const char*[1]){"ORDERS.*"};
cfg.SubjectsLen = 1;

// 创建一个流。如果您对返回的jsStreamInfo对象不感兴趣，
// 你可以传递NULL。
js_AddStream(NULL, js, &cfg, NULL, &jerr);

//更新一个流
cfg.MaxBytes = 8;
js_UpdateStream(NULL, js, &cfg, NULL, &jerr);

// 删除一个流
js_DeleteStream(js, "ORDERS", NULL, &jerr);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

KeyValue管理

如何创建密钥值存储的示例：

jsCtx       *js = NULL;
kvStore     *kv = NULL;
kvConfig    kvc;

// 假设我们在' js '中获得了一个JetStream上下文…

kvConfig_Init(&kvc);
kvc.Bucket = "KVS";
kvc.History = 10;
s = js_CreateKeyValue(&kv, js, &kvc);

// 做一些东西……

// 这是为了释放kv对象所使用的内存，
// 不删除服务器中的KeyValue存储
kvStore_Destroy(kv);

这显示了如何“绑定”到现有版本：

jsCtx       *js = NULL;
kvStore     *kv = NULL;

// 假设我们在' js '中获得了一个JetStream上下文…

s = js_KeyValue(&kv, ks, "KVS");

// 做一些东西……

// 这是为了释放kv对象所使用的内存，
// 不删除服务器中的KeyValue存储
kvStore_Destroy(kv);

以下是删除服务器中的 KeyValue 存储的方法：

jsCtx       *js = NULL;

// 假设我们在' js '中获得了一个JetStream上下文…

s = js_DeleteKeyValue(js, "KVS");
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

KeyValue API

这是如何为给定键输入值：

kvStore     *kv = NULL;
uint64_t    rev = 0;

// 假设我们有一个kvStore…

s = kvStore_PutString(&rev, kv, "MY_KEY", "my value");

// 如果一个不关心获取修订，传递NULL:
s = kvStore_PutString(NULL, kv, "MY_KEY", "my value");

上面为给定的键放置一个值，但是如果想要确保该值仅在以前从未存在过时才将其放置为该键，则可以调用：

// 与以前相同的注意事项:如果"rev"不需要，传递NULL:
s = kvStore_CreateString(&rev, kv, "MY_KEY", "my value");

当且仅当服务器中的最后一个修订版与传递给此 API 的修订版匹配时，才能更新密钥：

// 只有当这个键的当前版本(序列号)是10时，这将用值“我更新的值”更新键“MY_KEY”。
s = kvStore_UpdateString(&rev, kv, "MY_KEY", "my updated value", 10);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

以下是获取密钥的方法：

kvStore *kv = NULL;
kvEntry *e  = NULL;

// 假设我们有一个kvStore…
s = kvStore_Get(&e, kv, "MY_KEY");

// 然后我们可以从条目中获得一些字段:
printf("Key value: %s\n", kvEntry_ValueString(e));

// 一旦输入完成，我们需要销毁它以释放内存。
// 这不是从服务器删除密钥。
kvEntry_Destroy(e);
1
2
3
4
5
6
7
8
9
10
11
12

以下是清除密钥的方法：

kvStore *kv = NULL;

// 假设我们有一个kvStore…
s = kvStore_Purge(kv, "MY_KEY");

这将删除服务器中的密钥：

kvStore *kv = NULL;

// 假设我们有一个kvStore…
s = kvStore_Delete(kv, "MY_KEY");
1
2
3
4
5
6
7
8
9
10
11

要为给定密钥创建“观察程序”，请执行以下操作：

kvWatcher       *w = NULL;
kvWatchOptions  o;

// 假设我们有一个kvStore…

// 说我们对得到
// 删除标记……

// 初始化kvWatchOptions对象:
kvWatchOptions_Init(&o);
o.IgnoreDeletes = true;
// 创建观察者
s = kvStore_Watch(&w, kv, "foo.*", &o);
// 检查错误. .

// 现在更新:
while (some_condition)
{
    kvEntry *e = NULL;

    // 等待下一次更新最多5秒
    s = kvWatcher_Next(&e, w, 5000);

    // 对入口做点什么…

    // 销毁以释放内存
    kvEntry_Destroy(e);
}

// 当使用完观察者之后，它需要被销毁以释放内存:
kvWatcher_Destroy(w);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

要获取密钥的历史记录：

kvEntryList l;
int i;

// 列表在堆栈中定义，并将通过这个调用进行初始化/更新:

s = kvStore_History(&l, kv, "MY_KEY", NULL);

for (i=0; i<l.Count; i++)
{
    kvEntry *e = l.Entries[i];

    // 对入口做点什么…
}
// 当使用列表完成时，调用该函数来释放列表的条目和内容。
kvEntryList_Destroy(&l);

// 为了设置一个超时来获取历史记录，我们需要通过以下选项来实现:
kvWatchOptions o;

kvWatchOptions_Init(&o);
o.Timeout = 5000; // 5 seconds.
s = kvStore_History(&l, kv, "MY_KEY", &o);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

以下是获取存储桶密钥的方式：

kvKeysList  l;
int         i;

// 如果不需要任何选项，则传递NULL作为最后一个参数。
s = kvStore_Keys(&l, kv, NULL);
// 检查错误. .

// 检查所有键:
for (i=0; i<l.Count; i++)
    printf("Key: %s\n", l.Keys[i]);

// 完成后，名单需要销毁。
kvKeysList_Destroy(&l);

// 需要指定If选项:
kvWatchOptions o;

kvWatchOptions_Init(&o);
o.Timeout = 5000; // 5 seconds.
s = kvStore_Keys(&l, kv, &o);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

Headers

在版本 2.2.0+ 连接到服务器时，标头可用。

它们与 http 标头非常相似。它们是键/值对的映射，值是字符串数组。

标头允许用户添加有关消息的元信息，而不会干扰消息负载。

请注意，如果应用程序尝试在连接到不理解标头的服务器时发送带有标头的消息， 则发布调用将返回错误。NATS_NO_SERVER_SUPPORT

有一个 API 可以知道当前连接到的服务器是否支持标头：

natsStatus s = natsConnection_HasHeaderSupport(conn); if (s ==
NATS_NO_SERVER_SUPPORT)
// 处理服务器不支持此特性。 如果服务器理解标头，但即将将消息传递到不理解标头的客户端， 则标头将被剥离，以便较旧的客户端仍然可以接收消息。如果应用程序依赖于标头， 则将所有客户端和服务器都设置为支持标头的版本非常重要。

更多详情可前往https://github.com/nats-io/nats.c#nats-client下面进行查看。
更多API接口可前往文档http://nats-io.github.io/nats.c/查看

接下来我们可以测试一些简单的客户端程序来查看执行效果，但是再测试之前，我们还得去安装nats-server，得先启动nats服务器才能够进行测试。

NATS-SERVER

我们首先去下载一个nats-server，并且是能够直接使用的，可以去到https://github.com/nats-io/nats-server/releases/tag/v2.8.4进行下载。

下载之后进行解压，解压之后加入到目录里，然后通过命令./nats-server直接执行该nats服务器。

也可以使用命令./nats-server -D -P 4222 监听服务器

nats.c简单客户端测试示例

我们进入到getstarted目录里，以下是asyncsub.c源码

#include 

static void
onMsg(natsConnection *nc, natsSubscription *sub, natsMsg *msg, void *closure)
{
    printf("Received msg: %s - %.*s\n",
           natsMsg_GetSubject(msg),
           natsMsg_GetDataLength(msg),
           natsMsg_GetData(msg));

    // Need to destroy the message!
    natsMsg_Destroy(msg);

    // 通知主线程我们完成了。
    *(bool *)(closure) = true;
}

int main(int argc, char **argv)
{
    natsConnection      *conn = NULL;
    natsSubscription    *sub  = NULL;
    natsStatus          s;
    volatile bool       done  = false;

    printf("Listening on subject 'foo'\n");

    // 创建到默认nat URL的连接
    s = natsConnection_ConnectTo(&conn, NATS_DEFAULT_URL);
    if (s == NATS_OK)
    {
        // 在主题“foo”上创建异步订阅。
        // 当一个消息被发送到主题“foo”时，回调
        // onMsg()将被客户端库调用。
        // 你可以传递一个闭包作为最后一个参数。
        s = natsConnection_Subscribe(&sub, conn, "foo", onMsg, (void*) &done);
    }

    if (s == NATS_OK)
    {
        for (;!done;) {
            nats_Sleep(100);
        }
    }

    // 任何创造出来的东西都需要毁灭
    natsSubscription_Destroy(sub);
    natsConnection_Destroy(conn);

    // 如果出现错误，则打印堆栈跟踪并退出
    if (s != NATS_OK)
    {
        nats_PrintLastErrorStack(stderr);
        exit(2);
    }

    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

执行该文件，该可执行文件再我们之前cmake …的那个目录里，因为我之前是再build目录里进行创建的，所以我的那些可执行文件都在这些这个目录里

以下是pubstr.c文件

#include 

int main(int argc, char **argv)
{
    natsConnection      *conn = NULL;
    natsStatus          s;

    printf("Publishes a message on subject 'foo'\n");

    // 创建到默认nat URL的连接
    s = natsConnection_ConnectTo(&conn, NATS_DEFAULT_URL);
    if (s == NATS_OK)
    {
        // 这是一个方便的函数，可以在“foo”上发送消息
        // 作为一个字符串。
        s = natsConnection_PublishString(conn, "foo", "hello!");
    }

    // 任何创造出来的东西都需要毁灭
    natsConnection_Destroy(conn);

    // 如果出现错误，则打印堆栈跟踪并退出
    if (s != NATS_OK)
    {
        nats_PrintLastErrorStack(stderr);
        exit(2);
    }

    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

执行该文件,在asyncsub客户端那边可以得到由pubstr客户端这边发送过来的信息。

以上只是一个简单的示例，还有其他简单的示例，以及还有其他较为复杂的示例在examples目录里，感兴趣的可以去了解一下里面的源码及其工作原理，可以根据API接口文档进行查看。