Close consumer message channel when consumer closed.

**Is your feature request related to a problem? Please describe.**

lots of consumer code maybe like this:

```go
for msg := range consumer.Chan(){
// consume msg
}
```
if not close channel when consumer close, maybe cause goroutine leak.

**Describe the solution you'd like**

when close a consumer,close the consumer's message channel.

**Describe alternatives you've considered**

**Additional context**

Start from the consumer.Chan() message channel and trace the consumer close path. Check how channel ownership and shutdown are handled, then verify that closing a consumer causes ranging over consumer.Chan() to terminate without introducing sends to a closed channel.

Written by the indexing model from the issue text.