ssl生产消息 plaintext package main import ( "crypto/tls" "crypto/x509" "flag" "fmt" amqp "github.com/rabbitmq/amqp091go" "io/ioutil" "log" ) var ( uri flag.String("uri", "amqps://USERNAME:PASSWORD@10.10.33.196:5671", "AMQP URI") exchangeName flag.String("exchange", "goexchange", "Durable AMQP exchange name") exchangeType flag.String("exchangetype", "direct", "Exchange type directfanouttopicxcustom") routingKey flag.String("key", "testkey", "AMQP routing key") body flag.String("body", "foobar", "Body of message") reliable flag.Bool("reliable", true, "Wait for the publisher confirmation before exiting") ) func init() { flag.Parse() } func main() { if err : publish(uri, exchangeName, exchangeType, routingKey, body, reliable); err ! nil { log.Fatalf("%s", err) } log.Printf("published %dB OK", len(body)) } func publish(amqpsURI, exchange, exchangeType, routingKey, body string, reliable bool) error { caCert, err : ioutil.ReadFile("D:tmphzmqtest0520rabbitmqsslclientcacertificate.pem") if err ! nil { return err } cert, err : tls.LoadX509KeyPair("D:tmphzmqtest0520rabbitmqsslclientclientrabbitmqcertificate.pem", "D:tmphzmqtest0520rabbitmqsslclientclientrabbitmqkey.pem") if err ! nil { return err } rootCAs : x509.NewCertPool() rootCAs.AppendCertsFromPEM(caCert) tlsConf : &tls.Config{ RootCAs: rootCAs, Certificates: []tls.Certificate{cert}, //ServerName: "localhost", // Optional InsecureSkipVerify: true, } connection, err : amqp.DialTLS(amqpsURI, tlsConf) if err ! nil { return fmt.Errorf("Dial: %s", err) } defer connection.Close() log.Printf("got Connection, getting Channel") channel, err : connection.Channel() if err ! nil { return fmt.Errorf("Channel: %s", err) } log.Printf("got Channel, declaring %q Exchange (%q)", exchangeType, exchange) if err : channel.ExchangeDeclare( exchange, // name exchangeType, // type true, // durable false, // autodeleted false, // internal false, // noWait nil, // arguments ); err ! nil { return fmt.Errorf("Exchange Declare: %s", err) } // Reliable publisher confirms require confirm.select support from the // connection. if reliable { log.Printf("enabling publishing confirms.") if err : channel.Confirm(false); err ! nil { return fmt.Errorf("Channel could not be put into confirm mode: %s", err) } confirms : channel.NotifyPublish(make(chan amqp.Confirmation, 1)) defer confirmOne(confirms) } log.Printf("declared Exchange, publishing %dB body (%q)", len(body), body) if err channel.Publish( exchange, // publish to an exchange routingKey, // routing to 0 or more queues false, // mandatory false, // immediate amqp.Publishing{ Headers: amqp.Table{}, ContentType: "text/plain", ContentEncoding: "", Body: []byte(body), DeliveryMode: amqp.Transient, // 1nonpersistent, 2persistent Priority: 0, // 09 }, ); err ! nil { return fmt.Errorf("Exchange Publish: %s", err) } return nil } func confirmOne(confirms
.net 消费消息 plaintext using System; using System.Collections.Immutable; using System.Drawing; using Rabbit.Common.Data.Trades; using Rabbit.Common.Display; using RabbitMQ.Client; using RabbitMQ.Client.Events; namespace Rabbit.Example4.Consumer { internal sealed class Program { private static void Main() { var connectionFactory new ConnectionFactory { HostName "YOUR HOST IP", UserName "YOUR USER", Password "YOUR PASSWORD", Port 5672 }; using var connection connectionFactory.CreateConnection(); using var channel connection.CreateModel(); const string ExchangeName "dotnetexchange"; const string QueueName "dotnetqueue"; channel.ExchangeDeclare( exchange: ExchangeName, type: ExchangeType.Direct, durable: false, autoDelete: false, arguments: ImmutableDictionary .Empty); var queue channel.QueueDeclare( queue: QueueName, durable: false, exclusive: false, autoDelete: false, arguments: ImmutableDictionary .Empty); channel.QueueBind( queue: queue.QueueName, exchange: ExchangeName, routingKey: QueueName); var consumer new EventingBasicConsumer(channel); consumer.Received + (sender, eventArgs) > { var messageBody eventArgs.Body.ToArray(); var trade Trade.FromBytes(messageBody); DisplayInfo .For(trade) .SetExchange(eventArgs.Exchange) .SetQueue(queue.QueueName) .SetRoutingKey(eventArgs.RoutingKey) .SetVirtualHost(connectionFactory.VirtualHost) .Display(Color.Yellow); channel.BasicAck(eventArgs.DeliveryTag, multiple: false); }; channel.BasicConsume( queue: queue.QueueName, autoAck: false, consumer: consumer); Console.ReadLine(); } } } ssl生产消息 plaintext using System.Collections.Generic; using System.Collections.Immutable; using System.Drawing; using System.Threading.Tasks; using Rabbit.Common.Data.Trades; using Rabbit.Common.Display; using RabbitMQ.Client; namespace Rabbit.dotnet.Producer { internal sealed class Program { private static async Task Main() { var connectionFactory new ConnectionFactory { HostName "YOUR HOST IP", UserName "YOUR USERNAME", Password "YOUR PASSWORD", Port 5671 }; connectionFactory.AuthMechanisms new List () { new ExternalMechanismFactory() }; connectionFactory.Ssl.CertPath @"D:tmphzmqtest0520rabbitmqsslclientclientrabbitmqkey.p12"; // Ssl.CertPassphrase为固定参数,无需修改 connectionFactory.Ssl.CertPassphrase "YOUR PASSPHRASE"; connectionFactory.Ssl.Enabled true; connectionFactory.Ssl.CertificateValidationCallback (, , , ) > { return true; }; using var connection connectionFactory.CreateConnection(); using var channel connection.CreateModel(); const string ExchangeName "dotnetexchange"; const string QueueName "dotnetqueue"; channel.ExchangeDeclare( exchange: ExchangeName, type: ExchangeType.Direct, durable: false, autoDelete: false, arguments: ImmutableDictionary .Empty); var queue channel.QueueDeclare( queue: QueueName, durable: false, exclusive: false, autoDelete: false, arguments: ImmutableDictionary .Empty); channel.QueueBind( queue: queue.QueueName, exchange: ExchangeName, routingKey: QueueName, arguments: ImmutableDictionary .Empty); while (true) { var trade TradeData.GetFakeTrade(); string routingKey QueueName; channel.BasicPublish( exchange: ExchangeName, routingKey: routingKey, body: trade.ToBytes() ); DisplayInfo .For(trade) .SetExchange(ExchangeName) .SetRoutingKey(routingKey) .SetVirtualHost(connectionFactory.VirtualHost) .Display(Color.Cyan); await Task.Delay(millisecondsDelay: 1000); } } } }