mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-11-24 19:19:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

adjust publisher subscriber

Browse Source
This commit is contained in:
chrislu 2024-01-15 20:42:46 -08:00
parent 026c54a9bb
commit 7e6497cc1c
2 changed files with 10 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
weed/mq/client/cmd/weed_pub/publisher.go
View File

@ -12,7 +12,8 @@ import (
var ( var (
messageCount = flag.Int("n", 1000, "message count") messageCount = flag.Int("n", 1000, "message count")
concurrency = flag.Int("c", 4, "concurrency count") concurrency = flag.Int("c", 4, "concurrent publishers")
partitionCount = flag.Int("p", 6, "partition count")
namespace = flag.String("ns", "test", "namespace") namespace = flag.String("ns", "test", "namespace")
topic = flag.String("topic", "test", "topic") topic = flag.String("topic", "test", "topic")
@ -38,8 +39,8 @@ func doPublish(publisher *pub_client.TopicPublisher, id int) {
func main() { func main() {
flag.Parse() flag.Parse()
config := &pub_client.PublisherConfiguration{ config := &pub_client.PublisherConfiguration{
CreateTopic: true, CreateTopic: true,
CreateTopicPartitionCount: 1, CreateTopicPartitionCount: int32(*partitionCount),
} }
publisher := pub_client.NewTopicPublisher(*namespace, *topic, config) publisher := pub_client.NewTopicPublisher(*namespace, *topic, config)
brokers := strings.Split(*seedBrokers, ",") brokers := strings.Split(*seedBrokers, ",")

9
weed/mq/client/cmd/weed_sub/subscriber.go
View File

@ -5,6 +5,7 @@ import (
"fmt" "fmt"
"github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/mq/client/sub_client" "github.com/seaweedfs/seaweedfs/weed/mq/client/sub_client"
"github.com/seaweedfs/seaweedfs/weed/util"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/credentials/insecure"
"strings" "strings"
@ -15,15 +16,17 @@ var (
namespace = flag.String("ns", "test", "namespace") namespace = flag.String("ns", "test", "namespace")
topic = flag.String("topic", "test", "topic") topic = flag.String("topic", "test", "topic")
seedBrokers = flag.String("brokers", "localhost:17777", "seed brokers") seedBrokers = flag.String("brokers", "localhost:17777", "seed brokers")
clientId = flag.Uint("client_id", uint(util.RandomInt32()), "client id")
) )
func main() { func main() {
flag.Parse() flag.Parse()
subscriberConfig := &sub_client.SubscriberConfiguration{ subscriberConfig := &sub_client.SubscriberConfiguration{
ClientId: "testSubscriber", ClientId: fmt.Sprintf("client-%d", *clientId),
ConsumerGroup: "test", ConsumerGroup: "test",
ConsumerGroupInstanceId: "test", ConsumerGroupInstanceId: fmt.Sprintf("client-%d", *clientId),
GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
} }
@ -35,7 +38,7 @@ func main() {
} }
processorConfig := sub_client.ProcessorConfiguration{ processorConfig := sub_client.ProcessorConfiguration{
ConcurrentPartitionLimit: 6, ConcurrentPartitionLimit: 3,
} }
brokers := strings.Split(*seedBrokers, ",") brokers := strings.Split(*seedBrokers, ",")