youzan / nsq Goto Github PK

PublishOrdered(topic string, partitionKey []byte, body []byte)

Log file created at: 2017/10/30 16:26:25
Running on machine: 5d7074420ede
Binary: Built with gc go1.9.1 for linux/amd64
Log line format: [IWEF]mmdd hh:mm:ss.uuuuuu threadid file:line] msg

  ####### nslookup #######
    nsqlookupd:
      image: nsqio/nsq:v0.3.7-HA.1.6.0
      volumes:
        - ~/project/nsq/data:/data
      ports:
        - 4160:4160
        - 4161:4161
      command: /nsqlookupd -config /data/nsqlookupd.toml
      networks:
        - backend

  ####### nslookup #######
    nsqd:
      image: nsqio/nsq:v0.3.7-HA.1.6.0
      volumes:
        - ~/project/nsq/data:/data
      ports:
        - 4150:4150
        - 4151:4151
      command: /nsqd -config /data/nsqd.toml
      networks:
        - backend

  ####### nslookup #######
    nsqadmin:
      image: nsqio/nsq:v0.3.7-HA.1.6.0
      volumes:
        - ~/project/nsq/data:/data
      ports:
        - 4170:4170
        - 4171:4171
      command: /nsqadmin -config /data/nsqadmin.toml
      networks:
        - backend

## <addr>:<port> to listen on for TCP clients
tcp_address = "0.0.0.0:4160"

## <addr>:<port> to listen on for HTTP clients
http_address = "0.0.0.0:4161"

## rpc port used for cluster communication
rpc_port = "4260"
## address that will be registered with lookupd (defaults to the OS hostname)
#broadcast_address = ""
## the network interface for broadcast, the ip will be detected automatically.
# use this configure instead of broadcast_address to keep all the configure is the same 
broadcast_interface = "eth0"

## local reverse proxy port, basically used for collecting the stats 
# reverse_proxy_port = "4163"

cluster_id = "test-nsq-cluster-dev-1"
## the etcd cluster ip list
cluster_leadership_addresses = "http://etcd0:2379,http://etcd1:2379,http://etcd2:2379"

## duration of time a producer will remain in the active list since its last ping
inactive_producer_timeout = "100s"
# should at least twice as the ping interval on nsqd
nsqd_ping_timeout= "15s"

## duration of time a producer will remain tombstoned if registration remains
tombstone_lifetime = "45s"

## the detail of the log, larger number means more details
log_level = 2

## if empty, use the default flag value in glog
log_dir = "./"

## the time period (in hour) that the balance is allowed.
balance_interval = ["4", "5"]

## allow return topic as writable while no any channel under the topic
allow_write_with_nochannels = true

## unique identifier (int) for this worker (will default to a hash of hostname)
# id = 5150
#
## cluster id to separate different cluster
cluster_id = "test-nsq-cluster-dev-1"
## the etcd cluster ip list
cluster_leadership_addresses = "http://etcd0:2379,http://etcd1:2379,http://etcd2:2379"

## rpc port used for node communication for cluster , ip will be the same with broadcast
rpc_port = "4250"
## address that will be registered with lookupd (defaults to the OS hostname)
#broadcast_address = ""
broadcast_interface = "eth0"

## <addr>:<port> to listen on for TCP clients
tcp_address = "0.0.0.0:4150"

## <addr>:<port> to listen on for HTTP clients
http_address = "0.0.0.0:4151"

## <addr>:<port> to listen on for HTTPS clients
# https_address = "0.0.0.0:4152"

## local reverse proxy port, basically used for collecting the stats 
# reverse_proxy_port = "4153"

## keep alive heartbeat interval between the nsqd node with the nsqlookupd
lookup_ping_interval = "5s"

## path to store disk-backed messages
# data_path = "/var/lib/nsq"

## the remote message trace server
# remote_tracer = "127.0.0.1:1234"

## default retention days to keep the consumed topic data
retention_days = 7
## number of messages to keep in memory (per topic/channel)
mem_queue_size = 10000

## number of bytes per diskqueue file before rolling
max_bytes_per_file = 104857600

## number of messages per diskqueue fsync
sync_every = 2500

## duration of time per diskqueue fsync (time.Duration)
sync_timeout = "2s"


## duration to wait before auto-requeing a message
msg_timeout = "60s"

## maximum duration before a message will timeout
max_msg_timeout = "15m"

## maximum size of a single message in bytes
max_msg_size = 1024768

## maximum requeuing timeout for a message
max_req_timeout = "24h"

## duration threshold for requeue a message to the delayed queue end
req_to_end_threshold = "15m"

## maximum size of a single command body
max_body_size = 5123840

## maximum finished count with unordered
max_confirm_win = 5000

## maximum client configurable duration of time between client heartbeats
max_heartbeat_interval = "60s"

## maximum RDY count for a client
max_rdy_count = 2500

## maximum client configurable size (in bytes) for a client output buffer
max_output_buffer_size = 65536

## maximum client configurable duration of time between flushing to a client (time.Duration)
max_output_buffer_timeout = "1s"


## UDP <addr>:<port> of a statsd daemon for pushing stats
# statsd_address = "127.0.0.1:8125"

## prefix used for keys sent to statsd (%s for host replacement)
statsd_prefix = "nsq.%s"

## the protocol for statsd
statsd_protocol="tcp"

## duration between pushing to statsd (time.Duration)
statsd_interval = "60s"

## toggle sending memory and GC stats to statsd
statsd_mem_stats = true


## message processing time percentiles to keep track of (float)
e2e_processing_latency_percentiles = [
    100.0,
    99.0,
    95.0
]

## calculate end to end latency quantiles for this duration of time (time.Duration)
e2e_processing_latency_window_time = "10m"


## path to certificate file
tls_cert = ""

## path to private key file
tls_key = ""

## set policy on client certificate (require - client must provide certificate,
##  require-verify - client must provide verifiable signed certificate)
# tls_client_auth_policy = "require-verify"

## set custom root Certificate Authority
# tls_root_ca_file = ""

## require client TLS upgrades
tls_required = false

## minimum TLS version ("ssl3.0", "tls1.0," "tls1.1", "tls1.2")
tls_min_version = ""

## enable deflate feature negotiation (client compression)
deflate = true

## max deflate compression level a client can negotiate (> values == > nsqd CPU usage)
max_deflate_level = 6

## enable snappy feature negotiation (client compression)
snappy = true

## bigger level means more log details
log_level = 2

## if empty, use the default flag value in glog
log_dir = "./"

## the time period (in hour) that the auto clean is allowed.
auto_clean_interval = ["2", "4"]

## whether we should fix the data if only one ISR is available
start_as_fix_mode = true

## the interval for scan for channel timeout messages
queue_scan_interval = "100ms"
## selection channel count for each timeout scan 
queue_scan_selection_count = 100
## max timeout scan worker
queue_scan_worker_pool_max = 8
## re-scan trigger if there is dirty percent channels timeout
queue_scan_dirty_percent = 0.1

## <addr>:<port> to listen on for HTTP clients
http_address = ":4171"

## graphite HTTP address
graphite_url = ""

## proxy HTTP requests to graphite
proxy_graphite = false

## prefix used for keys sent to statsd (%s for host replacement, must match nsqd)
statsd_prefix = "nsq.%s"

## format of statsd counter stats
statsd_counter_format="%s"
statsd_gauge_format="%s"

## time interval nsqd is configured to push to statsd (must match nsqd)
statsd_interval = "60s"

## HTTP endpoint (fully qualified) to which POST notifications of admin actions will be sent
notification_http_endpoint = ""


## nsqlookupd HTTP addresses
nsqlookupd_http_addresses = [
    "http://nsqlookupd:4161"
]

## nsqd HTTP addresses (optional)
nsqd_http_addresses = [
    "http://nsqd:4151"
]

log_dir = "/data/logs/nsqadmin"

#trace_query_url = ""
#trace_app_id = ""
#trace_app_name = ""
#trace_log_index_id = ""
#trace_log_index_name = ""

I0506 13:30:47.069194   20696 router.go:273] httprouter.(*Router).recv ERROR: panic in HTTP handler - runtime error: invalid memory address or nil pointer dereference

package main

import (
	"fmt"
	"time"

	"github.com/youzan/go-nsq"
)

func main() {
	addr := "172.16.0.30:4161"

	config := nsq.NewConfig()
	config.MaxAttempts = 65535

	mgr, err := nsq.NewTopicProducerMgr([]string{"delay_test"}, config)
	if err != nil {
		panic(err)
	}

	mgr.ConnectToNSQLookupd(addr)
	err = mgr.Publish("delay_test", []byte("hello, "+time.Now().String()))
	if err != nil {
		panic(err)
	}

	// topic:delay_test 未开启order, 此处误操作，配置了 EnableOrdered 后续即便参数改回来，该 channel 的行为还是异常的
	config.EnableOrdered = true

	c, err := nsq.NewConsumer("delay_test", "default", config)
	if err != nil {
		panic(err)
	}

	c.AddConcurrentHandlers(nsq.HandlerFunc(func(msg *nsq.Message) error {
		fmt.Println(string(msg.Body), msg.Attempts)
		msg.RequeueWithoutBackoff(time.Second)
		return nil
	}), 1)

	err = c.ConnectToNSQLookupd(addr)
	if err != nil {
		panic(err)
	}

	select {}
}