export RABBITMQ_HOME="/home/lion/_app/rabbitmq_server-3.6.3" export PATH="$RABBITMQ_HOME/sbin:$PATH"

%% -*- mode: erlang -*- %% ---------------------------------------------------------------------------- %% RabbitMQ Sample Configuration File. %% %% See http://www.rabbitmq.com/configure.html for details. %% ---------------------------------------------------------------------------- [ {rabbit, [%% %% Network Connectivity %% ==================== %% %% By default, RabbitMQ will listen on all interfaces, using %% the standard (reserved) AMQP port. %% 默认的监听端口 %% {tcp_listeners, [5672]}, %% To listen on a specific interface, provide a tuple of {IpAddress, Port}. %% For example, to listen only on localhost for both IPv4 and IPv6: %% 也可以使用下面的格式进行指定IP和端口的监听 %% {tcp_listeners, [{"127.0.0.1", 5672}, %%                  {"::1",       5672}]}, %% SSL listeners are configured in the same fashion as TCP listeners, %% including the option to control the choice of interface. %% SSL连接端口配置 %% {ssl_listeners, [5671]}, %% Number of Erlang processes that will accept connections for the TCP %% and SSL listeners. %% TCP连接的进程数 %% {num_tcp_acceptors, 10}, %% {num_ssl_acceptors, 1}, %% Maximum time for AMQP 0-8/0-9/0-9-1 handshake (after socket connection %% and SSL handshake), in milliseconds. %% 超时时间，单位毫秒 %% {handshake_timeout, 10000}, %% Log levels (currently just used for connection logging). %% One of 'debug', 'info', 'warning', 'error' or 'none', in decreasing %% order of verbosity. Defaults to 'info'. %% 日志的级别，默认是info %% {log_levels, [{connection, info}, {channel, info}]}, %% Set to 'true' to perform reverse DNS lookups when accepting a %% connection. Hostnames will then be shown instead of IP addresses %% in rabbitmqctl and the management plugin. %% %% {reverse_dns_lookups, true}, %% %% Security / AAA %% ============== %% 安全配置 %% The default "guest" user is only permitted to access the server %% via a loopback interface (e.g. localhost). %% {loopback_users, [<<"guest">>]}, %%  %% Uncomment the following line if you want to allow access to the %% guest user from anywhere on the network. %% {loopback_users, []}, %% Configuring SSL. %% See http://www.rabbitmq.com/ssl.html for full documentation. %% %% {ssl_options, [{cacertfile,           "/path/to/testca/cacert.pem"}, %%                {certfile,             "/path/to/server/cert.pem"}, %%                {keyfile,              "/path/to/server/key.pem"}, %%                {verify,               verify_peer}, %%                {fail_if_no_peer_cert, false}]}, %% Choose the available SASL mechanism(s) to expose. %% The two default (built in) mechanisms are 'PLAIN' and %% 'AMQPLAIN'. Additional mechanisms can be added via %% plugins. %% %% See http://www.rabbitmq.com/authentication.html for more details. %% %% {auth_mechanisms, ['PLAIN', 'AMQPLAIN']}, %% Select an authentication database to use. RabbitMQ comes bundled %% with a built-in auth-database, based on mnesia. %% %% {auth_backends, [rabbit_auth_backend_internal]}, %% Configurations supporting the rabbitmq_auth_mechanism_ssl and %% rabbitmq_auth_backend_ldap plugins. %% %% NB: These options require that the relevant plugin is enabled. %% See http://www.rabbitmq.com/plugins.html for further details. %% The RabbitMQ-auth-mechanism-ssl plugin makes it possible to %% authenticate a user based on the client's SSL certificate. %% %% To use auth-mechanism-ssl, add to or replace the auth_mechanisms %% list with the entry 'EXTERNAL'. %% %% {auth_mechanisms, ['EXTERNAL']}, %% The rabbitmq_auth_backend_ldap plugin allows the broker to %% perform authentication and authorisation by deferring to an %% external LDAP server. %% %% For more information about configuring the LDAP backend, see %% http://www.rabbitmq.com/ldap.html. %% %% Enable the LDAP auth backend by adding to or replacing the %% auth_backends entry: %% %% {auth_backends, [rabbit_auth_backend_ldap]}, %% This pertains to both the rabbitmq_auth_mechanism_ssl plugin and %% STOMP ssl_cert_login configurations. See the rabbitmq_stomp %% configuration section later in this file and the README in %% https://github.com/rabbitmq/rabbitmq-auth-mechanism-ssl for further %% details. %% %% To use the SSL cert's CN instead of its DN as the username %% %% {ssl_cert_login_from, common_name}, %% SSL handshake timeout, in milliseconds. %% %% {ssl_handshake_timeout, 5000}, %% Password hashing implementation. Will only affect newly %% created users. To recalculate hash for an existing user %% it's necessary to update her password. %% %% {password_hashing_module, rabbit_password_hashing_sha256}, %% %% Default User / VHost %% ==================== %% 用户访问设置 %% On first start RabbitMQ will create a vhost and a user. These %% config items control what gets created. See %% http://www.rabbitmq.com/access-control.html for further %% information about vhosts and access control. %% %% {default_vhost,       <<"/">>}, %% {default_user,        <<"guest">>}, %% {default_pass,        <<"guest">>}, %% {default_permissions, [<<".*">>, <<".*">>, <<".*">>]}, %% Tags for default user %% %% For more details about tags, see the documentation for the %% Management Plugin at http://www.rabbitmq.com/management.html. %% %% {default_user_tags, [administrator]}, %% %% Additional network and protocol related configuration %% ===================================================== %% %% Set the default AMQP heartbeat delay (in seconds). %% 设置默认AMQP心跳延迟（秒） %% {heartbeat, 600}, %% Set the max permissible size of an AMQP frame (in bytes). %% %% {frame_max, 131072}, %% Set the max frame size the server will accept before connection %% tuning occurs %% %% {initial_frame_max, 4096}, %% Set the max permissible number of channels per connection. %% 0 means "no limit". %% %% {channel_max, 128}, %% Customising Socket Options. %% %% See (http://www.erlang.org/doc/man/inet.html#setopts-2) for %% further documentation. %% %% {tcp_listen_options, [{backlog,       128}, %%                       {nodelay,       true}, %%                       {exit_on_close, false}]}, %% %% Resource Limits & Flow Control %% ============================== %% %% See http://www.rabbitmq.com/memory.html for full details. %% Memory-based Flow Control threshold. %% %% {vm_memory_high_watermark, 0.4}, %% Alternatively, we can set a limit (in bytes) of RAM used by the node. %% %% {vm_memory_high_watermark, {absolute, 1073741824}}, %% %% Or you can set absolute value using memory units. %% %% {vm_memory_high_watermark, {absolute, "1024M"}}, %% %% Supported units suffixes: %% %% k, kiB: kibibytes (2^10 bytes) %% M, MiB: mebibytes (2^20) %% G, GiB: gibibytes (2^30) %% kB: kilobytes (10^3) %% MB: megabytes (10^6) %% GB: gigabytes (10^9) %% Fraction of the high watermark limit at which queues start to %% page message out to disc in order to free up memory. %% %% Values greater than 0.9 can be dangerous and should be used carefully. %% 内存最大使用比例 %% {vm_memory_high_watermark_paging_ratio, 0.5}, %% Interval (in milliseconds) at which we perform the check of the memory %% levels against the watermarks. %% 检查内存的间隔(毫秒) %% {memory_monitor_interval, 2500}, %% Set disk free limit (in bytes). Once free disk space reaches this %% lower bound, a disk alarm will be set - see the documentation %% listed above for more details. %% %% {disk_free_limit, 50000000}, %% %% Or you can set it using memory units (same as in vm_memory_high_watermark) %% {disk_free_limit, "50MB"}, %% {disk_free_limit, "50000kB"}, %% {disk_free_limit, "2GB"}, %% Alternatively, we can set a limit relative to total available RAM. %% %% Values lower than 1.0 can be dangerous and should be used carefully. %% {disk_free_limit, {mem_relative, 2.0}}, %% %% Misc/Advanced Options %% ===================== %% %% NB: Change these only if you understand what you are doing! %% %% To announce custom properties to clients on connection: %% %% {server_properties, []}, %% How to respond to cluster partitions. %% See http://www.rabbitmq.com/partitions.html for further details. %% %% {cluster_partition_handling, ignore}, %% Make clustering happen *automatically* at startup - only applied %% to nodes that have just been reset or started for the first time. %% See http://www.rabbitmq.com/clustering.html#auto-config for %% further details. %% 设置集群启动的节点 %% {cluster_nodes, {['rabbit@my.host.com'], disc}}, %% Interval (in milliseconds) at which we send keepalive messages %% to other cluster members. Note that this is not the same thing %% as net_ticktime; missed keepalive messages will not cause nodes %% to be considered down. %% 集群消息同步的时间(毫秒) %% {cluster_keepalive_interval, 10000}, %% Set (internal) statistics collection granularity. %% %% {collect_statistics, none}, %% Statistics collection interval (in milliseconds). %% %% {collect_statistics_interval, 5000}, %% Explicitly enable/disable hipe compilation. %% %% {hipe_compile, true}, %% Timeout used when waiting for Mnesia tables in a cluster to %% become available. %% %% {mnesia_table_loading_timeout, 30000}, %% Size in bytes below which to embed messages in the queue index. See %% http://www.rabbitmq.com/persistence-conf.html %% %% {queue_index_embed_msgs_below, 4096} ]}, %% ---------------------------------------------------------------------------- %% Advanced Erlang Networking/Clustering Options. %% %% See http://www.rabbitmq.com/clustering.html for details %% ---------------------------------------------------------------------------- {kernel, [%% Sets the net_kernel tick time. %% Please see http://erlang.org/doc/man/kernel_app.html and %% http://www.rabbitmq.com/nettick.html for further details. %% %% {net_ticktime, 60} ]}, %% ---------------------------------------------------------------------------- %% RabbitMQ Management Plugin %% %% See http://www.rabbitmq.com/management.html for details %% ---------------------------------------------------------------------------- {rabbitmq_management, [%% Pre-Load schema definitions from the following JSON file. See %% http://www.rabbitmq.com/management.html#load-definitions %% %% {load_definitions, "/path/to/schema.json"}, %% Log all requests to the management HTTP API to a file. %% 所有请求的HTTP API文件日志的路径。 %% {http_log_dir, "/path/to/access.log"}, %% Change the port on which the HTTP listener listens, %% specifying an interface for the web server to bind to. %% Also set the listener to use SSL and provide SSL options. %% Web管理的地址和端口 %% {listener, [{port,     12345}, %%             {ip,       "127.0.0.1"}, %%             {ssl,      true}, %%             {ssl_opts, [{cacertfile, "/path/to/cacert.pem"}, %%                         {certfile,   "/path/to/cert.pem"}, %%                         {keyfile,    "/path/to/key.pem"}]}]}, %% One of 'basic', 'detailed' or 'none'. See %% http://www.rabbitmq.com/management.html#fine-stats for more details. %% {rates_mode, basic}, %% Configure how long aggregated data (such as message rates and queue %% lengths) is retained. Please read the plugin's documentation in %% http://www.rabbitmq.com/management.html#configuration for more %% details. %% %% {sample_retention_policies, %%  [{global,   [{60, 5}, {3600, 60}, {86400, 1200}]}, %%   {basic,    [{60, 5}, {3600, 60}]}, %%   {detailed, [{10, 5}]}]} ]}, %% ---------------------------------------------------------------------------- %% RabbitMQ Shovel Plugin %% %% See http://www.rabbitmq.com/shovel.html for details %% ---------------------------------------------------------------------------- {rabbitmq_shovel, [{shovels, [%% A named shovel worker. %% {my_first_shovel, %%  [ %% List the source broker(s) from which to consume. %% %%   {sources, %%    [%% URI(s) and pre-declarations for all source broker(s). %%     {brokers, ["amqp://user:password@host.domain/my_vhost"]}, %%     {declarations, []} %%    ]}, %% List the destination broker(s) to publish to. %%   {destinations, %%    [%% A singular version of the 'brokers' element. %%     {broker, "amqp://"}, %%     {declarations, []} %%    ]}, %% Name of the queue to shovel messages from. %% %% {queue, <<"your-queue-name-goes-here">>}, %% Optional prefetch count. %% %% {prefetch_count, 10}, %% when to acknowledge messages: %% - no_ack: never (auto) %% - on_publish: after each message is republished %% - on_confirm: when the destination broker confirms receipt %% %% {ack_mode, on_confirm}, %% Overwrite fields of the outbound basic.publish. %% %% {publish_fields, [{exchange,    <<"my_exchange">>}, %%                   {routing_key, <<"from_shovel">>}]}, %% Static list of basic.properties to set on re-publication. %% %% {publish_properties, [{delivery_mode, 2}]}, %% The number of seconds to wait before attempting to %% reconnect in the event of a connection failure. %% %% {reconnect_delay, 2.5} %% ]} %% End of my_first_shovel ]} %% Rather than specifying some values per-shovel, you can specify %% them for all shovels here. %% %% {defaults, [{prefetch_count,     0}, %%             {ack_mode,           on_confirm}, %%             {publish_fields,     []}, %%             {publish_properties, [{delivery_mode, 2}]}, %%             {reconnect_delay,    2.5}]} ]}, %% ---------------------------------------------------------------------------- %% RabbitMQ Stomp Adapter %% %% See http://www.rabbitmq.com/stomp.html for details %% ---------------------------------------------------------------------------- {rabbitmq_stomp, [%% Network Configuration - the format is generally the same as for the broker %% Listen only on localhost (ipv4 & ipv6) on a specific port. %% {tcp_listeners, [{"127.0.0.1", 61613}, %%                  {"::1",       61613}]}, %% Listen for SSL connections on a specific port. %% {ssl_listeners, [61614]}, %% Number of Erlang processes that will accept connections for the TCP %% and SSL listeners. %% %% {num_tcp_acceptors, 10}, %% {num_ssl_acceptors, 1}, %% Additional SSL options %% Extract a name from the client's certificate when using SSL. %% %% {ssl_cert_login, true}, %% Set a default user name and password. This is used as the default login %% whenever a CONNECT frame omits the login and passcode headers. %% %% Please note that setting this will allow clients to connect without %% authenticating! %% %% {default_user, [{login,    "guest"}, %%                 {passcode, "guest"}]}, %% If a default user is configured, or you have configured use SSL client %% certificate based authentication, you can choose to allow clients to %% omit the CONNECT frame entirely. If set to true, the client is %% automatically connected as the default user or user supplied in the %% SSL certificate whenever the first frame sent on a session is not a %% CONNECT frame. %% %% {implicit_connect, true} ]}, %% ---------------------------------------------------------------------------- %% RabbitMQ MQTT Adapter %% %% See https://github.com/rabbitmq/rabbitmq-mqtt/blob/stable/README.md %% for details %% ---------------------------------------------------------------------------- {rabbitmq_mqtt, [%% Set the default user name and password. Will be used as the default login %% if a connecting client provides no other login details. %% %% Please note that setting this will allow clients to connect without %% authenticating! %% %% {default_user, <<"guest">>}, %% {default_pass, <<"guest">>}, %% Enable anonymous access. If this is set to false, clients MUST provide %% login information in order to connect. See the default_user/default_pass %% configuration elements for managing logins without authentication. %% %% {allow_anonymous, true}, %% If you have multiple chosts, specify the one to which the %% adapter connects. %% %% {vhost, <<"/">>}, %% Specify the exchange to which messages from MQTT clients are published. %% %% {exchange, <<"amq.topic">>}, %% Specify TTL (time to live) to control the lifetime of non-clean sessions. %% %% {subscription_ttl, 1800000}, %% Set the prefetch count (governing the maximum number of unacknowledged %% messages that will be delivered). %% %% {prefetch, 10}, %% TCP/SSL Configuration (as per the broker configuration). %% %% {tcp_listeners, [1883]}, %% {ssl_listeners, []}, %% Number of Erlang processes that will accept connections for the TCP %% and SSL listeners. %% %% {num_tcp_acceptors, 10}, %% {num_ssl_acceptors, 1}, %% TCP/Socket options (as per the broker configuration). %% %% {tcp_listen_options, [{backlog,   128}, %%                       {nodelay,   true}]} ]}, %% ---------------------------------------------------------------------------- %% RabbitMQ AMQP 1.0 Support %% %% See https://github.com/rabbitmq/rabbitmq-amqp1.0/blob/stable/README.md %% for details %% ---------------------------------------------------------------------------- {rabbitmq_amqp1_0, [%% Connections that are not authenticated with SASL will connect as this %% account. See the README for more information. %% %% Please note that setting this will allow clients to connect without %% authenticating! %% %% {default_user, "guest"}, %% Enable protocol strict mode. See the README for more information. %% %% {protocol_strict_mode, false} ]}, %% ---------------------------------------------------------------------------- %% RabbitMQ LDAP Plugin %% %% See http://www.rabbitmq.com/ldap.html for details. %% %% ---------------------------------------------------------------------------- {rabbitmq_auth_backend_ldap, [%% %% Connecting to the LDAP server(s) %% ================================ %% %% Specify servers to bind to. You *must* set this in order for the plugin %% to work properly. %% %% {servers, ["your-server-name-goes-here"]}, %% Connect to the LDAP server using SSL %% %% {use_ssl, false}, %% Specify the LDAP port to connect to %% %% {port, 389}, %% LDAP connection timeout, in milliseconds or 'infinity' %% %% {timeout, infinity}, %% Enable logging of LDAP queries. %% One of %%   - false (no logging is performed) %%   - true (verbose logging of the logic used by the plugin) %%   - network (as true, but additionally logs LDAP network traffic) %% %% Defaults to false. %% %% {log, false}, %% %% Authentication %% ============== %% %% Pattern to convert the username given through AMQP to a DN before %% binding %% %% {user_dn_pattern, "cn=${username},ou=People,dc=example,dc=com"}, %% Alternatively, you can convert a username to a Distinguished %% Name via an LDAP lookup after binding. See the documentation for %% full details. %% When converting a username to a dn via a lookup, set these to %% the name of the attribute that represents the user name, and the %% base DN for the lookup query. %% %% {dn_lookup_attribute,   "userPrincipalName"}, %% {dn_lookup_base,        "DC=gopivotal,DC=com"}, %% Controls how to bind for authorisation queries and also to %% retrieve the details of users logging in without presenting a %% password (e.g., SASL EXTERNAL). %% One of %%  - as_user (to bind as the authenticated user - requires a password) %%  - anon    (to bind anonymously) %%  - {UserDN, Password} (to bind with a specified user name and password) %% %% Defaults to 'as_user'. %% %% {other_bind, as_user}, %% %% Authorisation %% ============= %% %% The LDAP plugin can perform a variety of queries against your %% LDAP server to determine questions of authorisation. See %% http://www.rabbitmq.com/ldap.html#authorisation for more %% information. %% Set the query to use when determining vhost access %% %% {vhost_access_query, {in_group, %%                       "ou=${vhost}-users,ou=vhosts,dc=example,dc=com"}}, %% Set the query to use when determining resource (e.g., queue) access %% %% {resource_access_query, {constant, true}}, %% Set queries to determine which tags a user has %% %% {tag_queries, []} ]} ].

package main import ( "fmt" "log" "github.com/streadway/amqp" ) const ( //AMQP URI uri          = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange name exchangeName = "" //Durable AMQP queue name queueName    = "test-idoall-queues" //Body of message bodyMsg string = "hello idoall.org" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){ //调用发布消息函数 publish(uri, exchangeName, queueName, bodyMsg)
    log.Printf("published %dB OK", len(bodyMsg))
} //发布者的方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@queue, queue的名称 //@body, 主体内容 func publish(amqpURI string, exchange string, queue string, body string){ //建立连接 log.Printf("dialing %q", amqpURI)
    connection, err := amqp.Dial(amqpURI)
    failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
    channel, err := connection.Channel()
    failOnError(err, "Failed to open a channel") defer channel.Close()

    log.Printf("got queue, declaring %q", queue) //创建一个queue q, err := channel.QueueDeclare(
    queueName, // name false, // durable false, // delete when unused false, // exclusive false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue")

  log.Printf("declared queue, publishing %dB body (%q)", len(body), body) // Producer只能发送到exchange，它是不能直接发送到queue的。 // 现在我们使用默认的exchange（名字是空字符）。这个默认的exchange允许我们发送给指定的queue。 // routing_key就是指定的queue名字。 err = channel.Publish(
    exchange, // exchange q.Name, // routing key false, // mandatory false, // immediate amqp.Publishing {
      Headers:         amqp.Table{},
      ContentType: "text/plain",
      ContentEncoding: "",
      Body:        []byte(body),
    })
  failOnError(err, "Failed to publish a message")
}

package main import ( "fmt" "log" "github.com/streadway/amqp" ) const ( //AMQP URI uri           = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange nam exchangeName  = "" //Durable AMQP queue name queueName     = "test-idoall-queues" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){ //调用消息接收者 consumer(uri, exchangeName, queueName)
} //接收者方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@queue, queue的名称 func consumer(amqpURI string, exchange string, queue string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close()

  log.Printf("got queue, declaring %q", queue) //创建一个queue q, err := channel.QueueDeclare(
      queueName, // name false, // durable false, // delete when unused false, // exclusive false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue")

  log.Printf("Queue bound to Exchange, starting Consume") //订阅消息 msgs, err := channel.Consume(
      q.Name, // queue "", // consumer true, // auto-ack false, // exclusive false, // no-local false, // no-wait nil, // args )
  failOnError(err, "Failed to register a consumer") //创建一个channel forever := make(chan bool) //调用gorountine go func() { for d := range msgs {
        log.Printf("Received a message: %s", d.Body)
      }
  }()

  log.Printf(" [*] Waiting for messages. To exit press CTRL+C") //没有写入数据，一直等待读，阻塞当前线程，目的是让线程不退出 <-forever
}

package main import ( "fmt" "log" "os" "strings" "github.com/streadway/amqp" ) const ( //AMQP URI uri          = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange name exchangeName = "" //Durable AMQP queue name queueName    = "test-idoall-queues-task" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){
  bodyMsg := bodyFrom(os.Args) //调用发布消息函数 publish(uri, exchangeName, queueName, bodyMsg)
  log.Printf("published %dB OK", len(bodyMsg))
} func bodyFrom(args []string) string { var s string if (len(args) < 2) || os.Args[1] == "" {
                s = "hello idoall.org" } else {
                s = strings.Join(args[1:], " ")
        } return s
} //发布者的方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@queue, queue的名称 //@body, 主体内容 func publish(amqpURI string, exchange string, queue string, body string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close()

  log.Printf("got queue, declaring %q", queue) //创建一个queue q, err := channel.QueueDeclare(
    queueName, // name false, // durable false, // delete when unused false, // exclusive false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue")

  log.Printf("declared queue, publishing %dB body (%q)", len(body), body) // Producer只能发送到exchange，它是不能直接发送到queue的。 // 现在我们使用默认的exchange（名字是空字符）。这个默认的exchange允许我们发送给指定的queue。 // routing_key就是指定的queue名字。 err = channel.Publish(
    exchange, // exchange q.Name, // routing key false, // mandatory false, // immediate amqp.Publishing {
      Headers:         amqp.Table{},
      ContentType: "text/plain",
      ContentEncoding: "",
      Body:        []byte(body),
    })
  failOnError(err, "Failed to publish a message")
}

package main import ( "fmt" "log" "bytes" "time" "github.com/streadway/amqp" ) const ( //AMQP URI uri           = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange nam exchangeName  = "" //Durable AMQP queue name queueName     = "test-idoall-queues-task" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){ //调用消息接收者 consumer(uri, exchangeName, queueName)
} //接收者方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@queue, queue的名称 func consumer(amqpURI string, exchange string, queue string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close()

  log.Printf("got queue, declaring %q", queue) //创建一个queue q, err := channel.QueueDeclare(
      queueName, // name false, // durable false, // delete when unused false, // exclusive false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue")

  log.Printf("Queue bound to Exchange, starting Consume") //订阅消息 msgs, err := channel.Consume(
      q.Name, // queue "", // consumer false, // auto-ack false, // exclusive false, // no-local false, // no-wait nil, // args )
  failOnError(err, "Failed to register a consumer") //创建一个channel forever := make(chan bool) //调用gorountine go func() { for d := range msgs {
        log.Printf("Received a message: %s", d.Body)
        dot_count := bytes.Count(d.Body, []byte("."))
        t := time.Duration(dot_count)
        time.Sleep(t * time.Second)
        log.Printf("Done")
      }
  }()

  log.Printf(" [*] Waiting for messages. To exit press CTRL+C") //没有写入数据，一直等待读，阻塞当前线程，目的是让线程不退出 <-forever
}

package main import ( "fmt" "log" "os" "strings" "github.com/streadway/amqp" ) const ( //AMQP URI uri          = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange name exchangeName = "" //Durable AMQP queue name queueName    = "test-idoall-queues-acknowledgments" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){
  bodyMsg := bodyFrom(os.Args) //调用发布消息函数 publish(uri, exchangeName, queueName, bodyMsg)
  log.Printf("published %dB OK", len(bodyMsg))
} func bodyFrom(args []string) string { var s string if (len(args) < 2) || os.Args[1] == "" {
                s = "hello idoall.org" } else {
                s = strings.Join(args[1:], " ")
        } return s
} //发布者的方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@queue, queue的名称 //@body, 主体内容 func publish(amqpURI string, exchange string, queue string, body string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close()

  log.Printf("got queue, declaring %q", queue) //创建一个queue q, err := channel.QueueDeclare(
    queueName, // name false, // durable false, // delete when unused false, // exclusive false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue")

  log.Printf("declared queue, publishing %dB body (%q)", len(body), body) // Producer只能发送到exchange，它是不能直接发送到queue的。 // 现在我们使用默认的exchange（名字是空字符）。这个默认的exchange允许我们发送给指定的queue。 // routing_key就是指定的queue名字。 err = channel.Publish(
    exchange, // exchange q.Name, // routing key false, // mandatory false, // immediate amqp.Publishing {
      Headers:         amqp.Table{},
      ContentType: "text/plain",
      ContentEncoding: "",
      Body:        []byte(body),
    })
  failOnError(err, "Failed to publish a message")
}

package main import ( "fmt" "log" "bytes" "time" "github.com/streadway/amqp" ) const ( //AMQP URI uri           = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange nam exchangeName  = "" //Durable AMQP queue name queueName     = "test-idoall-queues-acknowledgments" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){ //调用消息接收者 consumer(uri, exchangeName, queueName)
} //接收者方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@queue, queue的名称 func consumer(amqpURI string, exchange string, queue string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close()

  log.Printf("got queue, declaring %q", queue) //创建一个queue q, err := channel.QueueDeclare(
      queueName, // name false, // durable false, // delete when unused false, // exclusive false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue")

  log.Printf("Queue bound to Exchange, starting Consume") //订阅消息 msgs, err := channel.Consume(
      q.Name, // queue "", // consumer false, // auto-ack false, // exclusive false, // no-local false, // no-wait nil, // args )
  failOnError(err, "Failed to register a consumer") //创建一个channel forever := make(chan bool) //调用gorountine go func() { for d := range msgs {
        log.Printf("Received a message: %s", d.Body)
        dot_count := bytes.Count(d.Body, []byte("."))
        t := time.Duration(dot_count)
        time.Sleep(t * time.Second)
        log.Printf("Done")
        d.Ack(false)
      }
  }()

  log.Printf(" [*] Waiting for messages. To exit press CTRL+C") //没有写入数据，一直等待读，阻塞当前线程，目的是让线程不退出 <-forever
}

package main import ( "fmt" "log" "os" "strings" "github.com/streadway/amqp" ) const ( //AMQP URI uri          = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange name exchangeName = "" //Durable AMQP queue name queueName    = "test-idoall-queues-durability" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){
  bodyMsg := bodyFrom(os.Args) //调用发布消息函数 publish(uri, exchangeName, queueName, bodyMsg)
  log.Printf("published %dB OK", len(bodyMsg))
} func bodyFrom(args []string) string { var s string if (len(args) < 2) || os.Args[1] == "" {
                s = "hello idoall.org" } else {
                s = strings.Join(args[1:], " ")
        } return s
} //发布者的方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@queue, queue的名称 //@body, 主体内容 func publish(amqpURI string, exchange string, queue string, body string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close()

  log.Printf("got queue, declaring %q", queue) //创建一个queue q, err := channel.QueueDeclare(
    queueName, // name true, // durable false, // delete when unused false, // exclusive false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue")

  log.Printf("declared queue, publishing %dB body (%q)", len(body), body) // Producer只能发送到exchange，它是不能直接发送到queue的。 // 现在我们使用默认的exchange（名字是空字符）。这个默认的exchange允许我们发送给指定的queue。 // routing_key就是指定的queue名字。 err = channel.Publish(
    exchange, // exchange q.Name, // routing key false, // mandatory false, // immediate amqp.Publishing {
      Headers:         amqp.Table{},
      DeliveryMode: amqp.Persistent,
      ContentType: "text/plain",
      ContentEncoding: "",
      Body:        []byte(body),
    })
  failOnError(err, "Failed to publish a message")
}

package main import ( "fmt" "log" "bytes" "time" "github.com/streadway/amqp" ) const ( //AMQP URI uri           = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange nam exchangeName  = "" //Durable AMQP queue name queueName     = "test-idoall-queues-durability" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){ //调用消息接收者 consumer(uri, exchangeName, queueName)
} //接收者方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@queue, queue的名称 func consumer(amqpURI string, exchange string, queue string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close()

  log.Printf("got queue, declaring %q", queue) //创建一个queue q, err := channel.QueueDeclare(
      queueName, // name true, // durable false, // delete when unused false, // exclusive false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue")

  log.Printf("Queue bound to Exchange, starting Consume") //订阅消息 msgs, err := channel.Consume(
      q.Name, // queue "", // consumer false, // auto-ack false, // exclusive false, // no-local false, // no-wait nil, // args )
  failOnError(err, "Failed to register a consumer") //创建一个channel forever := make(chan bool) //调用gorountine go func() { for d := range msgs {
        log.Printf("Received a message: %s", d.Body)
        dot_count := bytes.Count(d.Body, []byte("."))
        t := time.Duration(dot_count)
        time.Sleep(t * time.Second)
        log.Printf("Done")
        d.Ack(false)
      }
  }()

  log.Printf(" [*] Waiting for messages. To exit press CTRL+C") //没有写入数据，一直等待读，阻塞当前线程，目的是让线程不退出 <-forever
}

package main import ( "fmt" "log" "os" "strings" "github.com/streadway/amqp" ) const ( //AMQP URI uri          = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange name exchangeName = "" //Durable AMQP queue name queueName    = "test-idoall-queues-fair_dispatch" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){
  bodyMsg := bodyFrom(os.Args) //调用发布消息函数 publish(uri, exchangeName, queueName, bodyMsg)
  log.Printf("published %dB OK", len(bodyMsg))
} func bodyFrom(args []string) string { var s string if (len(args) < 2) || os.Args[1] == "" {
                s = "hello idoall.org" } else {
                s = strings.Join(args[1:], " ")
        } return s
} //发布者的方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@queue, queue的名称 //@body, 主体内容 func publish(amqpURI string, exchange string, queue string, body string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close()

  log.Printf("got queue, declaring %q", queue) //创建一个queue q, err := channel.QueueDeclare(
    queueName, // name true, // durable false, // delete when unused false, // exclusive false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue")

  log.Printf("declared queue, publishing %dB body (%q)", len(body), body) // Producer只能发送到exchange，它是不能直接发送到queue的。 // 现在我们使用默认的exchange（名字是空字符）。这个默认的exchange允许我们发送给指定的queue。 // routing_key就是指定的queue名字。 err = channel.Publish(
    exchange, // exchange q.Name, // routing key false, // mandatory false, // immediate amqp.Publishing {
      Headers:         amqp.Table{},
      DeliveryMode: amqp.Persistent,
      ContentType: "text/plain",
      ContentEncoding: "",
      Body:        []byte(body),
    })
  failOnError(err, "Failed to publish a message")
}

package main import ( "fmt" "log" "bytes" "time" "github.com/streadway/amqp" ) const ( //AMQP URI uri           = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange nam exchangeName  = "" //Durable AMQP queue name queueName     = "test-idoall-queues-fair_dispatch" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){ //调用消息接收者 consumer(uri, exchangeName, queueName)
} //接收者方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@queue, queue的名称 func consumer(amqpURI string, exchange string, queue string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close()

  log.Printf("got queue, declaring %q", queue) //创建一个queue q, err := channel.QueueDeclare(
      queueName, // name true, // durable false, // delete when unused false, // exclusive false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue") //每次只取一条消息 err = channel.Qos( 1, // prefetch count 0, // prefetch size false, // global )
  failOnError(err, "Failed to set QoS")

  log.Printf("Queue bound to Exchange, starting Consume") //订阅消息 msgs, err := channel.Consume(
      q.Name, // queue "", // consumer false, // auto-ack false, // exclusive false, // no-local false, // no-wait nil, // args )
  failOnError(err, "Failed to register a consumer") //创建一个channel forever := make(chan bool) //调用gorountine go func() { for d := range msgs {
        log.Printf("Received a message: %s", d.Body)
        dot_count := bytes.Count(d.Body, []byte("."))
        t := time.Duration(dot_count)
        time.Sleep(t * time.Second)
        log.Printf("Done")
        d.Ack(false)
      }
  }()

  log.Printf(" [*] Waiting for messages. To exit press CTRL+C") //没有写入数据，一直等待读，阻塞当前线程，目的是让线程不退出 <-forever
}

package main import ( "fmt" "log" "os" "strings" "github.com/streadway/amqp" ) const ( //AMQP URI uri          = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange name exchangeName = "test-idoall-exchange-logs" //Exchange type - direct|fanout|topic|x-custom exchangeType = "fanout" //AMQP routing key routingKey   = "" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){
  bodyMsg := bodyFrom(os.Args) //调用发布消息函数 publish(uri, exchangeName, exchangeType, routingKey, bodyMsg)
  log.Printf("published %dB OK", len(bodyMsg))
} func bodyFrom(args []string) string { var s string if (len(args) < 2) || os.Args[1] == "" {
                s = "hello idoall.org" } else {
                s = strings.Join(args[1:], " ")
        } return s
} //发布者的方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@exchangeType, exchangeType的类型direct|fanout|topic //@routingKey, routingKey的名称 //@body, 主体内容 func publish(amqpURI string, exchange string, exchangeType string, routingKey string, body string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close() //创建一个queue log.Printf("got Channel, declaring %q Exchange (%q)", exchangeType, exchange)
  err = channel.ExchangeDeclare(
    exchange, // name exchangeType, // type true, // durable false, // auto-deleted false, // internal false, // noWait nil, // arguments )
  failOnError(err, "Failed to declare a queue") // 发布消息 log.Printf("declared queue, publishing %dB body (%q)", len(body), body)
  err = channel.Publish(
    exchange, // exchange routingKey, // routing key false, // mandatory false, // immediate amqp.Publishing {
      Headers:         amqp.Table{},
      ContentType: "text/plain",
      ContentEncoding: "",
      Body:        []byte(body),
    })
  failOnError(err, "Failed to publish a message")
}

package main import ( "fmt" "log" "github.com/streadway/amqp" ) const ( //AMQP URI uri           = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange name exchangeName = "test-idoall-exchange-logs" //Exchange type - direct|fanout|topic|x-custom exchangeType = "fanout" //AMQP binding key bindingKey   = "" //Durable AMQP queue name queueName     = "" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){ //调用消息接收者 consumer(uri, exchangeName, exchangeType, queueName, bindingKey)
} //接收者方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@exchangeType, exchangeType的类型direct|fanout|topic //@queue, queue的名称 //@key , 绑定的key名称 func consumer(amqpURI string, exchange string, exchangeType string, queue string, key string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close() //创建一个exchange log.Printf("got Channel, declaring Exchange (%q)", exchange)
  err = channel.ExchangeDeclare(
    exchange, // name of the exchange exchangeType, // type true, // durable false, // delete when complete false, // internal false, // noWait nil, // arguments );
  failOnError(err, "Exchange Declare:") //创建一个queue q, err := channel.QueueDeclare(
      queueName, // name false, // durable false, // delete when unused true, // exclusive 当Consumer关闭连接时，这个queue要被deleted  false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue") //绑定到exchange err = channel.QueueBind(
    q.Name, // name of the queue key, // bindingKey exchange, // sourceExchange false, // noWait nil, // arguments );
  failOnError(err, "Failed to bind a queue")

  log.Printf("Queue bound to Exchange, starting Consume") //订阅消息 msgs, err := channel.Consume(
      q.Name, // queue "", // consumer false, // auto-ack false, // exclusive false, // no-local false, // no-wait nil, // args )
  failOnError(err, "Failed to register a consumer") //创建一个channel forever := make(chan bool) //调用gorountine go func() { for d := range msgs {
        log.Printf(" [x] %s", d.Body)
      }
  }()

  log.Printf(" [*] Waiting for messages. To exit press CTRL+C") //没有写入数据，一直等待读，阻塞当前线程，目的是让线程不退出 <-forever
}

package main import ( "fmt" "log" "os" "strings" "github.com/streadway/amqp" ) const ( //AMQP URI uri          = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange name exchangeName = "test-idoall-exchange-direct-logs" //Exchange type - direct|fanout|topic|x-custom exchangeType = "direct" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){
  bodyMsg := bodyFrom(os.Args) //调用发布消息函数 publish(uri, exchangeName, exchangeType, bodyMsg)
  log.Printf("published %dB OK", len(bodyMsg))
} func bodyFrom(args []string) string { var s string if (len(args) < 3) || os.Args[2] == "" {
                s = "hello idoall.org" } else {
                s = strings.Join(args[2:], " ")
        } return s
} func severityFrom(args []string) string { var s string if (len(args) < 2) || os.Args[1] == "" {
                s = "info" } else {
                s = os.Args[1]
        } return s
} //发布者的方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@exchangeType, exchangeType的类型direct|fanout|topic //@body, 主体内容 func publish(amqpURI string, exchange string, exchangeType string, body string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close() //创建一个queue log.Printf("got Channel, declaring %q Exchange (%q)", exchangeType, exchange)
  err = channel.ExchangeDeclare(
    exchange, // name exchangeType, // type true, // durable false, // auto-deleted false, // internal false, // noWait nil, // arguments )
  failOnError(err, "Failed to declare a queue") // 发布消息 log.Printf("declared queue, publishing %dB body (%q)", len(body), body)
  err = channel.Publish(
    exchange, // exchange severityFrom(os.Args), // routing key false, // mandatory false, // immediate amqp.Publishing {
      Headers:         amqp.Table{},
      ContentType: "text/plain",
      ContentEncoding: "",
      Body:        []byte(body),
    })
  failOnError(err, "Failed to publish a message")
}

package main import ( "fmt" "log" "os" "github.com/streadway/amqp" ) const ( //AMQP URI uri           = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange name exchangeName = "test-idoall-exchange-direct-logs" //Exchange type - direct|fanout|topic|x-custom exchangeType = "direct" //AMQP binding key bindingKey   = "" //Durable AMQP queue name queueName     = "" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){ //调用消息接收者 consumer(uri, exchangeName, exchangeType, queueName, bindingKey)
} //接收者方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@exchangeType, exchangeType的类型direct|fanout|topic //@queue, queue的名称 //@key , 绑定的key名称 func consumer(amqpURI string, exchange string, exchangeType string, queue string, key string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close() //创建一个exchange log.Printf("got Channel, declaring Exchange (%q)", exchange)
  err = channel.ExchangeDeclare(
    exchange, // name of the exchange exchangeType, // type true, // durable false, // delete when complete false, // internal false, // noWait nil, // arguments );
  failOnError(err, "Exchange Declare:") //创建一个queue q, err := channel.QueueDeclare(
      queueName, // name false, // durable false, // delete when unused true, // exclusive 当Consumer关闭连接时，这个queue要被deleted  false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue") if len(os.Args) < 2 {
          log.Printf("Usage: %s [info] [warning] [error]", os.Args[0])
          os.Exit(0)
  } for _, s := range os.Args[1:] {
          log.Printf("Binding queue %s to exchange %s with routing key %s",
                  q.Name, exchange, s) //绑定到exchange err = channel.QueueBind(
            q.Name, // name of the queue s, // bindingKey exchange, // sourceExchange false, // noWait nil, // arguments );
          failOnError(err, "Failed to bind a queue")
  }

  log.Printf("Queue bound to Exchange, starting Consume") //订阅消息 msgs, err := channel.Consume(
      q.Name, // queue "", // consumer false, // auto-ack false, // exclusive false, // no-local false, // no-wait nil, // args )
  failOnError(err, "Failed to register a consumer") //创建一个channel forever := make(chan bool) //调用gorountine go func() { for d := range msgs {
        log.Printf(" [x] %s", d.Body)
      }
  }()

  log.Printf(" [*] Waiting for messages. To exit press CTRL+C") //没有写入数据，一直等待读，阻塞当前线程，目的是让线程不退出 <-forever
}

package main import ( "fmt" "log" "os" "strings" "github.com/streadway/amqp" ) const ( //AMQP URI uri          = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange name exchangeName = "test-idoall-exchange-direct-logs" //Exchange type - direct|fanout|topic|x-custom exchangeType = "fanout" //AMQP routing key routingKey   = "" //Durable AMQP queue name queueName    = "test-idoall-queues-direct" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){
  bodyMsg := bodyFrom(os.Args) //调用发布消息函数 publish(uri, exchangeName, exchangeType, routingKey, bodyMsg)
  log.Printf("published %dB OK", len(bodyMsg))
} func bodyFrom(args []string) string { var s string if (len(args) < 2) || os.Args[1] == "" {
                s = "hello idoall.org" } else {
                s = strings.Join(args[1:], " ")
        } return s
} //发布者的方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@exchangeType, exchangeType的类型direct|fanout|topic //@routingKey, routingKey的名称 //@body, 主体内容 func publish(amqpURI string, exchange string, exchangeType string, routingKey string, body string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close() //创建一个queue log.Printf("got Channel, declaring %q Exchange (%q)", exchangeType, exchange)
  err = channel.ExchangeDeclare(
    exchange, // name exchangeType, // type true, // durable false, // auto-deleted false, // internal false, // noWait nil, // arguments )
  failOnError(err, "Failed to declare a queue") // 发布消息 log.Printf("declared queue, publishing %dB body (%q)", len(body), body)
  err = channel.Publish(
    exchange, // exchange routingKey, // routing key false, // mandatory false, // immediate amqp.Publishing {
      Headers:         amqp.Table{},
      ContentType: "text/plain",
      ContentEncoding: "",
      Body:        []byte(body),
    })
  failOnError(err, "Failed to publish a message")
}

package main import ( "fmt" "log" "github.com/streadway/amqp" ) const ( //AMQP URI uri           = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange name exchangeName = "test-idoall-exchange-topic-logs" //Exchange type - direct|fanout|topic|x-custom exchangeType = "topic" //AMQP binding key bindingKey   = "" //Durable AMQP queue name queueName     = "" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){ //调用消息接收者 consumer(uri, exchangeName, exchangeType, queueName, bindingKey)
} //接收者方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@exchangeType, exchangeType的类型direct|fanout|topic //@queue, queue的名称 //@key , 绑定的key名称 func consumer(amqpURI string, exchange string, exchangeType string, queue string, key string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close() //创建一个exchange log.Printf("got Channel, declaring Exchange (%q)", exchange)
  err = channel.ExchangeDeclare(
    exchange, // name of the exchange exchangeType, // type true, // durable false, // delete when complete false, // internal false, // noWait nil, // arguments );
  failOnError(err, "Exchange Declare:") //创建一个queue q, err := channel.QueueDeclare(
      queueName, // name false, // durable false, // delete when unused true, // exclusive 当Consumer关闭连接时，这个queue要被deleted  false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue") //绑定到exchange err = channel.QueueBind(
    q.Name, // name of the queue key, // bindingKey exchange, // sourceExchange false, // noWait nil, // arguments );
  failOnError(err, "Failed to bind a queue")

  log.Printf("Queue bound to Exchange, starting Consume") //订阅消息 msgs, err := channel.Consume(
      q.Name, // queue "", // consumer false, // auto-ack false, // exclusive false, // no-local false, // no-wait nil, // args )
  failOnError(err, "Failed to register a consumer") //创建一个channel forever := make(chan bool) //调用gorountine go func() { for d := range msgs {
        log.Printf(" [x] %s", d.Body)
      }
  }()

  log.Printf(" [*] Waiting for messages. To exit press CTRL+C") //没有写入数据，一直等待读，阻塞当前线程，目的是让线程不退出 <-forever
}

package main import ( "fmt" "log" "strconv" "github.com/streadway/amqp" ) const ( //AMQP URI uri          = "amqp://guest:guest@localhost:5672/" //Durable AMQP queue name queueName = "rpc-queue" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func main(){ //调用发布消息函数 publish(uri, queueName)
} //发布者的方法 // //@amqpURI, amqp的地址 //@queue, queue的名称 func publish(amqpURI string, queue string){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close() //创建一个queue log.Printf("got queue, declaring %q", queue)
  q,err := channel.QueueDeclare(
          queue, // name false, // durable false, // delete when usused false, // exclusive false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue") //均衡处理，每次处理一条消息 err = channel.Qos( 1, // prefetch count 0, // prefetch size false, // global )
  failOnError(err, "Failed to set QoS") //订阅一个消息 //log.Printf("Queue bound to Exchange, starting Consume") msgs, err := channel.Consume(
          q.Name, // queue "", // consumer false, // auto-ack false, // exclusive false, // no-local false, // no-wait nil, // args )
  failOnError(err, "Failed to register a consumer")

  forever := make(chan bool) // 发布消息 go func() { for d := range msgs {
                  n, err := strconv.Atoi(string(d.Body))
                  failOnError(err, "Failed to convert body to integer")

                  log.Printf(" [.] server端接收到的数据是 (%d)", n)
                  response := n*2 err = channel.Publish( "", // exchange d.ReplyTo, // routing key false, // mandatory false, // immediate amqp.Publishing{
                                  ContentType: "text/plain",
                                  CorrelationId: d.CorrelationId,
                                  Body:          []byte(strconv.Itoa(response)),
                          })
                  failOnError(err, "Failed to publish a message")

                  d.Ack(false)
          }
  }()

  log.Printf(" [*] Awaiting RPC requests") //没有写入数据，一直等待读，阻塞当前线程，目的是让线程不退出 <-forever
}

package main import ( "fmt" "log" "math/rand" "os" "strconv" "strings" "time" "github.com/streadway/amqp" ) const ( //AMQP URI uri           = "amqp://guest:guest@localhost:5672/" //Durable AMQP exchange name exchangeName = "" //Exchange type - direct|fanout|topic|x-custom queueName = "rpc-queue" ) //如果存在错误，则输出 func failOnError(err error, msg string) { if err != nil {
    log.Fatalf("%s: %s", msg, err) panic(fmt.Sprintf("%s: %s", msg, err))
  }
} func randomString(l int) string {
        bytes := make([]byte, l) for i := 0; i < l; i++ {
                bytes[i] = byte(randInt(65, 90))
        } return string(bytes)
} func randInt(min int, max int) int { return min + rand.Intn(max-min)
} func bodyFrom(args []string) int { var s string if (len(args) < 2) || os.Args[1] == "" {
                s = "30" } else {
                s = strings.Join(args[1:], " ")
        }
        n, err := strconv.Atoi(s)
        failOnError(err, "Failed to convert arg to integer") return n
} func main(){

    rand.Seed(time.Now().UTC().UnixNano())

    n := bodyFrom(os.Args)

    log.Printf(" [x] 请求的数据是(%d)", n)
    res, err := fibonacciRPC(n, uri, exchangeName, queueName)
    failOnError(err, "Failed to handle RPC request")

    log.Printf(" [.] 计算结果为 %d", res)
} //RPC client调用方法 // //@amqpURI, amqp的地址 //@exchange, exchange的名称 //@queue, queue的名称 func fibonacciRPC(n int, amqpURI string, exchange string, queue string) (res int, err error){ //建立连接 log.Printf("dialing %q", amqpURI)
  connection, err := amqp.Dial(amqpURI)
  failOnError(err, "Failed to connect to RabbitMQ") defer connection.Close() //创建一个Channel log.Printf("got Connection, getting Channel")
  channel, err := connection.Channel()
  failOnError(err, "Failed to open a channel") defer channel.Close() //创建一个queue log.Printf("got queue, declaring %q", queue)
  q,err := channel.QueueDeclare( "", // name false, // durable false, // delete when usused true, // exclusive false, // no-wait nil, // arguments )
  failOnError(err, "Failed to declare a queue")

  log.Printf("Queue bound to Exchange, starting Consume") //订阅消息 msgs, err := channel.Consume(
      q.Name, // queue "", // consumer true, // auto-ack false, // exclusive false, // no-local false, // no-wait nil, // args )
  failOnError(err, "Failed to register a consumer")

  corrId := randomString(32)

  err = channel.Publish( "", // exchange queue, // routing key false, // mandatory false, // immediate amqp.Publishing{
                  ContentType: "text/plain",
                  CorrelationId: corrId,
                  ReplyTo:       q.Name,
                  Body:          []byte(strconv.Itoa(n)),
          })
  failOnError(err, "Failed to publish a message") for d := range msgs { if corrId == d.CorrelationId {
                  res, err = strconv.Atoi(string(d.Body))
                  failOnError(err, "Failed to convert body to integer") break }
  } return }