标签: channels

有没有办法在没有实际尝试发送的情况下测试go频道发送的失败？标准的非阻塞发送是这样的:

msg := "hi"
select {
    case messages <- msg:
        fmt.Println("sent message", msg)
    default:
        fmt.Println("no message sent")
    }

问题是我需要准备好发送"msg"来测试频道.

我想测试一下发送是否会以不需要"msg"准备发送的方式失败.

我有在docker容器中运行的django应用程序.最近我发现我需要在我的应用程序中添加websockets接口.我在nginx和redis后面使用带有daphne的通道作为缓存.问题是我必须在1个容器中运行django worker和daphne.在容器启动时运行的脚本:

#!/usr/bin/env bash

python wait_for_postgres.py
python manage.py makemigrations
python manage.py migrate
python manage.py collectstatic --no-input

python manage.py runworker --only-channels=http.* --only-channels=websocket.* -v2
daphne team_up.asgi:channel_layer --port 8000 -b 0.0.0.0

但它依赖于经营一名工人.我试过nohup,但它似乎不起作用.如果我直接从带有docker exec的容器运行daphne一切正常.

在某些 Windows 10 机器上使用 Django Channels 时出现问题。基本上我无法连接到套接字 [错误 10061] 并且 python 给了我一个错误

ERROR - server - Exception inside application: Multiple exceptions: [Errno 10061] Connect call failed ('::1', 6379), [Errno 10061] Connect call failed ('127. 0.0.1', 6379)

我知道这是 Windows/OS 级别的问题。我已经关闭了所有防火墙等。仍然无法连接到套接字

示例存储库：https : //github.com/andrewgodwin/channels-examples

我试图找到使用工作程序生成的结果的正确方法，同时在所有工作完成后优雅地退出结果循环。为了说明，我做了下面的例子。我的实际情况与此示例略有不同，因为我不知道每个工作程序 go 例程将返回多少“工作”，显然这些 for 循环正在执行固定数量的结果 (5)。

我是 goroutines 和 channel 的新手，但以下是我理解的基本租户；

• 只有发件人应该关闭频道
• range在通道上执行将继续直到通道关闭

package main

import (
    "fmt"
    "sync"
)

func worker1(r chan string, wg *sync.WaitGroup) {
    for i := 0; i < 5; i++ {
        r <- fmt.Sprintf("1.%d", i)
    }

    wg.Done()
}

func worker2(r chan string, wg *sync.WaitGroup) {
    for i := 0; i < 5; i++ {
        r <- fmt.Sprintf("2.%d", i)
    }

    wg.Done()
}

func main() {
    var wg sync.WaitGroup

    r := make(chan string)
    wg.Add(2)
    go worker1(r, …

这是 Go 中的 Concurrency 一书中的代码示例。在 select 块中是以下语句

case takeStream <- <- valueStream:

看不懂双箭头是干什么的，文中也没有解释。当我将其替换为时，输出会发生变化

case takeStream <- valueStream:

所以这显然是必要的

功能齐全：

func take(done<- chan interface{}, valueStream <- chan interface{}, num int) <- chan interface{}{
    takeStream := make ( chan interface{})
    go func() {
        defer close(takeStream)
        for i := 0; i < num; i ++ {
            select {
            case <- done :
                return
            case takeStream <- <- valueStream:
            }
        }
    }()

    return takeStream
}

编辑：如果我理解正确，扩展出来的语句将是

i := 5
valueStream <- i
tmp <- …

我在Uber 的风格指南上读到，一个人最多应该使用 1 的通道长度。

虽然我很清楚使用 100 或 1000 的通道大小是非常糟糕的做法，但我想知道为什么通道大小为 10 不被视为有效选项。我错过了一些部分来得出正确的结论。

在下面，您可以遵循由一些基准测试支持的我的论点（和反论点）。

我知道，如果您负责从该通道写入或读取的两个 go-routines 将在顺序写入或读取到/从通道之间被其他一些 IO 操作中断，则预计不会从更高的通道获得收益缓冲，我同意 1 是最好的选择。

但是，可以说，除了由通道写入/读取引起的隐式锁定和解锁之外，不需要其他重要的 go-routine 切换。那么我总结如下：

在使用大小为 1 和 10 的通道缓冲区（GR = go-routine）的通道上处理 100 个值时，考虑上下文切换的数量

• 缓冲区=1：（GR1 插入 1 个值，GR2 读取 1 个值）X 100 ~ 200 个 go-routine 切换
• Buffer=10：（GR1 插入 10 个值，GR2 读取 10 个值）X 10 ~ 20 个 go-routine 开关

我做了一些基准测试来证明这实际上速度更快：

package main

import (
    "testing"
)

type a struct {
    b [100]int64
}

func BenchmarkBuffer1(b *testing.B) {
    count …

我不明白double {}的含义.这在任何学习材料中都没有明确说明.谢谢.

variable <-struct {}{} 

附加的要点是一个使用生产者/多消费者模型中的渠道的简单程序.由于某些原因,

go run channels.go 打印所有结果但不返回(并且没有死锁或至少不会让我发生死锁的恐慌.)

type walkietalkie struct {
    in chan int
    out chan int
    quit chan bool
}

var items []int = []int{
    0, 1, 2, 3, 4, 5,
}

func work1(q walkietalkie) {
    for {
        select {
        case a, more := <- q.in:
            if more {
                q.out <- a * 2
            }
        default:
            break
        }
    }
}

func work2(q walkietalkie) {
    for  {
        select {
        case a, more := <- q.in:
            if more {
                q.out <- a * …

使用 daphne 这是我的设置：

简介：

 web: daphne my_application.asgi:application --port $PORT --bind 0.0.0.0 -v2

设置

 INSTALLED_APPS = [
        'django.contrib.admin',
        'django.contrib.auth',
        'django.contrib.contenttypes',
        'django.contrib.sessions',
        'django.contrib.messages',
        'django.contrib.staticfiles',
        'django.contrib.humanize',
        'channels',
        'django_summernote',
        ....
    ]

     CHANNEL_LAYERS = {
        "default": {
            "BACKEND": "channels_redis.core.RedisChannelLayer",
            "CONFIG": {
                "hosts": [os.environ.get('REDIS_URL', 'redis://localhost:6379')],
            }
        }
    } 

    ASGI_APPLICATION = "my_application.routing.application"

路由文件：

    from channels.auth import AuthMiddlewareStack
    from channels.routing import ProtocolTypeRouter, URLRouter
    from django.conf.urls import url

    application = ProtocolTypeRouter({
        'websocket': AuthMiddlewareStack(
            URLRouter(
                [
                    url(*),
                    ....
                ]
            )
        ),
    })

ASGI.PY - 错误发生的地方

  """
    ASGI entrypoint. Configures Django …

我有一个消耗通道元素的例程。这些元素在任意时间可用（通道连接到网络套接字，不同的客户端在不同的时间提供输入），并且需要通过速率受限的 API 推送，该 API 可以一次处理成批的元素。

我目前的解决方案是使用一个列表容器和一个自动收报机：一个 goroutine 从通道中抓取元素并将它们推送到一个列表中，然后一个会触发每个“不受限制的最短时间”的自动收报机获取列表​​的内容并将其推送到限速 API。

我想出了这个（我很抱歉，我不是 Go 程序员）：

// simulate the channel
c := make(chan int, 100)
go func() {
    for i := 1; i < 30000; i++ {
        c <- i
        time.Sleep(1 * time.Millisecond)
    }
}()

// block-read the channel
l := []int{}
var m sync.Mutex
go func(list *[]int) {
    for val := range c {
        m.Lock()
        *list = append(*list, val)
        m.Unlock()
    }
}(l)

// read batches every tick
for range time.Tick(1 * time.Second) { …

我有这段代码。

func Start() bool {
    for {
        if checkSomthingIsTrue() {
            if err := doSomthing(); err != nil {
                continue
            }
        }
        select {
        case <-ctx.Done():
            return true
        }
    }
}

如何在不使用default:case 的情况下使上述函数非阻塞。不使用默认情况的原因是因为它总是占用 100% 的 CPU。

答：我已经使用 time.Ticker 来限制了 谢谢