由于nats-streaming废弃，nats使用jetstream可以存储消息，下面使用java和go分别测试jetstream
1、maven:
<dependency>
<groupId>io.nats</groupId>
<artifactId>jnats</artifactId>
<version>2.13.2</version>
</dependency>
2、测试代码：
先运行go run jspub.go，再运行go run jssub.go
1、go代码 jetstream
注意几点：
//1、AddStream pub和sub不同
//pub部分
//stream StreamConfig的Subjects不支持通配符//Subjects: []string{"ORDERS.*"},
js.AddStream(&nats.StreamConfig{
Name: "ORDERS",
Subjects: []string{"ORDERS.scratch"},
//Subjects: []string{"ORDERS.*"},//jetstream不支持通配符
Retention: nats.WorkQueuePolicy,
})
//sub部分
//stream StreamConfig的Subjects可以使用//Subjects: []string{"ORDERS.*"},
js.AddStream(&nats.StreamConfig{
Name: "ORDERS",
Subjects: []string{"ORDERS.scratch"},
//可以Subjects: []string{"ORDERS.*"},
Retention: nats.WorkQueuePolicy,
})
//2、三种订阅（选第一或第二种即可）
/**
第一种：js.Subscribe、这2种方式都可以"ORDERS.*" "ORDERS.scratch"
第二种：js.SubscribeSync可以使用("ORDERS.scratch") 不能使用("ORDERS.*")
第三种：PullSubscribe无法订阅消息
*/
1、go jetstream pub
package main
import (
"encoding/json"
"fmt"
"github.com/nats-io/nats.go"
"runtime"
"strconv"
"time"
)
func main() {
// Connect to NATS
nc, _ := nats.Connect(nats.DefaultURL)
// Create JetStream Context
js, _ := nc.JetStream(nats.PublishAsyncMaxPending(256))
//js, _ := nc.JetStream()
js.DeleteConsumer("ORDERS", "MONITOR")
js.DeleteStream("ORDERS")
js.AddStream(&nats.StreamConfig{
Name: "ORDERS",
Subjects: []string{"ORDERS.scratch"},
//Subjects: []string{"ORDERS.*"},//jetstream不支持通配符
Retention: nats.WorkQueuePolicy,
})
js.UpdateStream(&nats.StreamConfig{
Name: "ORDERS",
MaxBytes: 8,
})
js.AddConsumer("ORDERS", &nats.ConsumerConfig{ //存消息
Durable: "MONITOR",
})
//打印信息
info, _ := js.StreamInfo("ORDERS")
marshal, _ := json.Marshal(info)
fmt.Println("===> StreamInfo ", string(marshal))
consumerInfo, _ := js.ConsumerInfo("ORDERS", "MONITOR")
marshal2, _ := json.Marshal(consumerInfo)
fmt.Println("===> ConsumerInfo ", string(marshal2))
// Simple Stream Publisher
js.Publish("ORDERS.scratch", []byte("hello"))
// Simple Async Stream Publisher
max := 5000
for i := 0; i < max; i++ {
js.PublishAsync("ORDERS.scratch", []byte("hello "+strconv.Itoa(i)))
time.Sleep(time.Duration(500) * time.Millisecond)
}
runtime.Goexit()
}
2、go jetstream sub
package main
import (
"encoding/json"
"fmt"
"github.com/nats-io/nats.go"
"log"
"runtime"
)
func printnatsinfo(js nats.JetStreamContext) {
info, _ := js.StreamInfo("ORDERS")
marshal, _ := json.Marshal(info)
fmt.Println("===> StreamInfo ", string(marshal))
consumerInfo, _ := js.ConsumerInfo("ORDERS", "MONITOR")
marshal2, _ := json.Marshal(consumerInfo)
fmt.Println("===> ConsumerInfo ", string(marshal2))
}
func main() {
nc, err := nats.Connect(nats.DefaultURL)
if err != nil {
log.Fatal("connect error")
}
js, _ := nc.JetStream()
printnatsinfo(js)
//sub端如何delstream的话 消息会丢失
//js.DeleteConsumer("ORDERS", "MONITOR")
//js.DeleteStream("ORDERS")
js.AddStream(&nats.StreamConfig{
Name: "ORDERS",
Subjects: []string{"ORDERS.scratch"},
//Subjects: []string{"ORDERS.*"}, //jetstream不支持通配符
Retention: nats.WorkQueuePolicy,
})
js.UpdateStream(&nats.StreamConfig{
Name: "ORDERS",
MaxBytes: 8,
})
js.AddConsumer("ORDERS", &nats.ConsumerConfig{
Durable: "MONITOR",
})
printnatsinfo(js)
//2、三种订阅（选第一或第二种即可）
/**
第一种：js.Subscribe、这2种方式都可以"ORDERS.*" "ORDERS.scratch"
第二种：js.SubscribeSync可以使用("ORDERS.scratch") 不能使用("ORDERS.*")
第三种：PullSubscribe无法订阅消息
*/
//第一种订阅
//js.Subscribe、这2种方式都可以"ORDERS.*" "ORDERS.scratch"
js.Subscribe("ORDERS.*", func(m *nats.Msg) {
//js.Subscribe("ORDERS.scratch", func(m *nats.Msg) {
fmt.Printf("Received a JetStream message: %s\n", string(m.Data))
m.Ack()
})
//第二种订阅
//js.SubscribeSync可以使用("ORDERS.scratch") 不能使用("ORDERS.*")
// Simple Sync Durable Consumer (optional SubOpts at the end)
/*sub, _ := js.SubscribeSync("ORDERS.scratch", nats.Durable("MONITOR"), nats.MaxDeliver(3))
for {
m, _ := sub.NextMsg(3 * time.Second)
if m != nil {
fmt.Printf("<=== Received a JetStream message: %s\n", string(m.Data))
m.Ack()
}
}*/
//第三种订阅
/**
PullSubscribe无法订阅消息
*/
/*subscribe, err := js.PullSubscribe("ORDERS.scratch", "MONITOR")
for {
msgs, _ := subscribe.Fetch(3)
fmt.Println("<=== ", msgs)
for i, x := range msgs {
fmt.Printf("第 %d 位 x 的值 = %d\n", i, x)
}
time.Sleep(time.Duration(3) * time.Second)
}*/
runtime.Goexit()
}
2、java代码 pub/sub jetstream
import io.nats.client.*;
import io.nats.client.api.*;
import lombok.extern.slf4j.Slf4j;
import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
@Slf4j
public class NatsJetStreamTest {
private Connection nc;
private JetStream jetStream;
@Before
public void pre() throws InterruptedException {
Options options = new Options.Builder().server("nats://127.0.0.1:4222"))
.connectionListener(getConnectionListener()).build();
Nats.connectAsynchronously(options, true);
}
private ConnectionListener getConnectionListener() {
return (conn, type) -> {
try {
log.info("nats event {}.", type);
if (type == ConnectionListener.Events.CONNECTED) {
nc = conn;
String subject = this.getChannelSubjectName(null);
String stream = this.getChannelStreamName(null);
this.delConsumer(stream);
this.delStream(stream);
this.addChannelStream(null);
ConsumerConfiguration cc = ConsumerConfiguration.builder().maxAckPending(1).durable("MONITOR").build();
PushSubscribeOptions so = PushSubscribeOptions.builder().stream(stream).configuration(cc).build();
jetStream = nc.jetStream();
Dispatcher dispatcher = nc.createDispatcher();
/*jetStream.subscribe(subject, dispatcher, (msg) -> {
log.info("Nats receive room Message {}", msg);
msg.ack();
}, false, so);*/
log.info("初始化完成，可以发送数据");
}
} catch (Exception e) {
log.error("ConnectionListener error", e);
}
};
}
@Test
public void jetstreamtest() throws InterruptedException, JetStreamApiException, IOException {
while (jetStream == null) {
Thread.sleep(1000);
}
log.info("开始发送数据");
int max = 500;
for (int i = 0; i < max; i++) {
jetStream.publish(this.getChannelSubjectName(null), ("hello === " + i).getBytes(StandardCharsets.UTF_8));
Thread.sleep(1000);
}
}
public String getChannelStreamName(String channelId) {
return "ORDERS";
}
public String getChannelSubjectName(String channelId) {
return "ORDERS.scratch";
}
public void publishToStream(String subject, String message) {
try {
nc.jetStream().publish(subject, message.getBytes(StandardCharsets.UTF_8));
} catch (IOException | JetStreamApiException e) {
log.error("nats publishToStream error", e);
}
}
public static StreamInfo getStreamInfoOrNullWhenNotExist(JetStreamManagement jsm, String streamName) throws IOException, JetStreamApiException {
try {
return jsm.getStreamInfo(streamName);
}
catch (JetStreamApiException jsae) {
if (jsae.getErrorCode() == 404) {
return null;
}
throw jsae;
}
}
public static void printStreamInfo(StreamInfo si) {
printObject(si, "StreamConfiguration", "StreamState", "ClusterInfo", "Mirror", "subjects", "sources");
}
public static void printObject(Object o, String... subObjectNames) {
String s = o.toString();
for (String sub : subObjectNames) {
boolean noIndent = sub.startsWith("!");
String sb = noIndent ? sub.substring(1) : sub;
String rx1 = ", " + sb;
String repl1 = (noIndent ? ",\n": ",\n ") + sb;
s = s.replace(rx1, repl1);
}
log.info(s);
}
public void addChannelStream(String channelId) {
try {
String channelSubject = getChannelSubjectName(channelId);
String channelStream = getChannelStreamName(channelId);
StreamInfo streamInfo = this.createOrGetChannelStream(channelStream);
StreamConfiguration configuration = streamInfo.getConfiguration();
List<String> subjects = configuration.getSubjects();
if (Ts.hv(subjects) && subjects.contains(channelSubject)) {
return;
}
StreamConfiguration streamConfig = StreamConfiguration.builder(configuration).addSubjects(channelSubject).build();
nc.jetStreamManagement().updateStream(streamConfig);
} catch (IOException | JetStreamApiException e) {
log.error("nats addSubjectToStream error", e);
}
}
public void delStream(String channelStream) throws IOException, JetStreamApiException {
JetStreamManagement jsm = nc.jetStreamManagement();
try {
jsm.deleteStream(channelStream);
} catch (JetStreamApiException e) {
if (e.getErrorCode() != 404) {
throw e;
}
}
}
public StreamInfo createOrGetChannelStream(String streamName) throws IOException, JetStreamApiException {
JetStreamManagement jsm = nc.jetStreamManagement();
StreamInfo streamInfo = getStreamInfoOrNullWhenNotExist(jsm, streamName);
if (!Ts.hv(streamInfo)) {
StreamConfiguration streamConfig = StreamConfiguration.builder().name(streamName)
// .subjects(subject1)
.retentionPolicy(RetentionPolicy.WorkQueue)
// .maxConsumers(...)
// .maxBytes(...)
.maxAge(Duration.ofHours(1))
// .maxMsgSize(...)
.storageType(StorageType.Memory)
// .replicas(...)
// .noAck(...)
// .template(...)
// .discardPolicy(...)
.build();
streamInfo = jsm.addStream(streamConfig);
}
printStreamInfo(streamInfo);
return streamInfo;
}
public void delConsumer(String channelStream) throws IOException, JetStreamApiException {
JetStreamManagement jsm = nc.jetStreamManagement();
try {
List<String> consumerNames = jsm.getConsumerNames(channelStream);
if (Ts.hv(consumerNames)) {
log.info("Nats stream {} has consumers need del.", channelStream);
for (String consumerName : consumerNames) {
jsm.deleteConsumer(channelStream, consumerName);
}
}
} catch (JetStreamApiException e) {
if (e.getErrorCode() != 404) {
throw e;
}
}
}
}
网友评论