SDK de inicio rápido de Streaming para Go
Publique y consuma mensajes en el servicio Streaming mediante el SDK de OCI para Go.
En este inicio rápido se muestra cómo utilizar el SDK de Oracle Cloud Infrastructure (OCI) para Go y Oracle Cloud Infrastructure Streaming para publicar y consumir mensajes.
Para conocer los conceptos clave y más detalles de Streaming, consulte Visión general de Streaming. Para obtener más información sobre el uso de los SDK de OCI, consulte las Guías sobre SDK.
Requisitos
-
Para utilizar el SDK para Go, debe tener lo siguiente:
- Una cuenta de Oracle Cloud Infrastructure.
- Un usuario creado en esa cuenta, en un grupo con una política que otorgue los permisos necesarios. Este usuario puede ser usted mismo u otra persona/sistema que necesite llamar a la API. Para obtener un ejemplo de cómo configurar un nuevo usuario, grupo, compartimento y política, consulte Adición de usuarios. Para obtener una lista de las políticas típicas que puede que desee utilizar, consulte Políticas Comunes.
- Par de claves utilizado para firmar solicitudes de API con la clave pública cargada en Oracle. Solo el usuario que llama a la API debe poseer la clave privada. Para obtener más información, consulte la sección sobre el archivo de configuración de SDK.
- Recopile el punto final de mensajes y el OCID de un flujo. Para obtener más información sobre un flujo, consulte Obtención de detalles de un flujo. Para este inicio rápido, el flujo debe utilizar un punto final público y permitir que Oracle gestione el cifrado. Consulte Creación de un flujo y Creación de un pool de flujos si no tiene un flujo existente.
- Go instalado de forma local. Siga estas instrucciones si es necesario. Asegúrese de que
go
esté enPATH
. -
Visual Studio Code (recomendado) o cualquier otro entorno de desarrollo integrado (IDE) o editor de texto.
- Asegúrese de que tiene un archivo de configuración de SDK válido. Para entornos de producción, debe utilizar la autorización de principal de instancia.
Producción de mensajes
- Abra su editor favorito, como Visual Studio Code, desde el directorio de trabajo vacío
wd
. - Cree un archivo denominado
Producer.go
en este directorio. -
Agregue el siguiente código a
Producer.go
. Sustituya los valores de las variablesociConfigFilePath
,ociProfileName
,ociStreamOcid
yociMessageEndpoint
en el siguiente fragmento de código por los valores aplicables a su arrendamiento.package main import ( "context" "fmt" "strconv" "github.com/oracle/oci-go-sdk/v36/common" "github.com/oracle/oci-go-sdk/v36/example/helpers" "github.com/oracle/oci-go-sdk/v36/streaming" ) const ociMessageEndpoint = "<stream_message_endpoint>" const ociStreamOcid = "<stream_OCID>" const ociConfigFilePath = "<config_file_path>" const ociProfileName = "<config_file_profile_name>" func main() { fmt.Println("Go oci oss sdk example producer") putMsgInStream(ociMessageEndpoint, ociStreamOcid) } func putMsgInStream(streamEndpoint string, streamOcid string) { fmt.Println("Stream endpoint for put msg api is: " + streamEndpoint) provider, err := common.ConfigurationProviderFromFileWithProfile(ociConfigFilePath, ociProfileName, "") helpers.FatalIfError(err) streamClient, err := streaming.NewStreamClientWithConfigurationProvider(provider, streamEndpoint) helpers.FatalIfError(err) // Create a request and dependent object(s). for i := 0; i < 5; i++ { putMsgReq := streaming.PutMessagesRequest{StreamId: common.String(streamOcid), PutMessagesDetails: streaming.PutMessagesDetails{ // we are batching 2 messages for each Put Request Messages: []streaming.PutMessagesDetailsEntry{ {Key: []byte("key dummy-0-" + strconv.Itoa(i)), Value: []byte("value dummy-" + strconv.Itoa(i))}, {Key: []byte("key dummy-1-" + strconv.Itoa(i)), Value: []byte("value dummy-" + strconv.Itoa(i))}}}, } // Send the request using the service client putMsgResp, err := streamClient.PutMessages(context.Background(), putMsgReq) helpers.FatalIfError(err) // Retrieve value from the response. fmt.Println(putMsgResp) } }
- Guarde
Producer.go
. -
Abra el terminal, utilice
cd
para acceder al directoriowd
y ejecute los siguientes comandos en orden:-
Este comando crea el archivo
go.mod
en el directoriowd
:go mod init oss_producer_example/v0
-
Este comando instala el SDK de OCI para Go y para Streaming:
go mod tidy
-
Este comando ejecuta el ejemplo:
go run Producer.go
-
- Mostrar los últimos mensajes enviados al flujo para ver los últimos mensajes enviados al flujo para verificar que la producción se ha realizado correctamente.
Consumo de mensajes
- En primer lugar, asegúrese de que el flujo del que desea consumir mensajes contiene mensajes. Puede utilizar la consola para producir un mensaje de prueba o utilizar el flujo y los mensajes que hemos creado en este inicio rápido.
-
Agregue el siguiente código a
Consumer.go
. Sustituya los valores de las variablesociConfigFilePath
,ociProfileName
,ociStreamOcid
yociMessageEndpoint
en el siguiente fragmento de código por los valores aplicables a su arrendamiento.package main import ( "context" "fmt" "github.com/oracle/oci-go-sdk/v36/common" "github.com/oracle/oci-go-sdk/v36/example/helpers" "github.com/oracle/oci-go-sdk/v36/streaming" ) const ociMessageEndpoint = "<stream_message_endpoint>" const ociStreamOcid = "<stream_OCID>" const ociConfigFilePath = "<config_file_path>" const ociProfileName = "<config_file_profile_name>" func main() { fmt.Println("Go oci oss sdk example for consumer") getMsgWithGroupCursor(ociMessageEndpoint, ociStreamOcid) } func getMsgWithGroupCursor(streamEndpoint string, streamOcid string) { client, err := streaming.NewStreamClientWithConfigurationProvider(common.DefaultConfigProvider(), streamEndpoint) helpers.FatalIfError(err) grpCursorCreateReq0 := streaming.CreateGroupCursorRequest{ StreamId: common.String(streamOcid), CreateGroupCursorDetails: streaming.CreateGroupCursorDetails{Type: streaming.CreateGroupCursorDetailsTypeTrimHorizon, CommitOnGet: common.Bool(true), GroupName: common.String("Go-groupname-0"), InstanceName: common.String("Go-groupname-0-instancename-0"), TimeoutInMs: common.Int(1000), }} // Send the request using the service client grpCursorResp0, err := client.CreateGroupCursor(context.Background(), grpCursorCreateReq0) helpers.FatalIfError(err) // Retrieve value from the response. fmt.Println(grpCursorResp0) simpleGetMsgLoop(client, streamOcid, *grpCursorResp0.Value) } func simpleGetMsgLoop(streamClient streaming.StreamClient, streamOcid string, cursorValue string) { for i := 0; i < 5; i++ { getMsgReq := streaming.GetMessagesRequest{Limit: common.Int(3), StreamId: common.String(streamOcid), Cursor: common.String(cursorValue)} // Send the request using the service client getMsgResp, err := streamClient.GetMessages(context.Background(), getMsgReq) helpers.FatalIfError(err) // Retrieve value from the response. if len(getMsgResp.Items) > 0 { fmt.Println("Key : " + string(getMsgResp.Items[0].Key) + ", value : " + string(getMsgResp.Items[0].Value) + ", Partition " + *getMsgResp.Items[0].Partition) } if len(getMsgResp.Items) > 1 { fmt.Println("Key : " + string(getMsgResp.Items[1].Key) + ", value : " + string(getMsgResp.Items[1].Value) + ", Partition " + *getMsgResp.Items[1].Partition) } cursorValue = *getMsgResp.OpcNextCursor } }
- Guarde
Consumer.go
. -
Abra el terminal, utilice
cd
para acceder al directoriowd
y ejecute los siguientes comandos en orden:-
Este comando crea el archivo
go.mod
en el directoriowd
:go mod init oss_consumer_example/v0
-
Este comando instala el SDK de OCI para Go y para Streaming:
go mod tidy
-
Este comando ejecuta el ejemplo:
go run Consumer.go
-
-
Se mostrarán mensajes similares a los siguientes:
Go oci oss sdk example for consumer { RawResponse={200 OK 200 HTTP/1.1 1 1 map[Access-Control-Allow-Credentials:[true] ... } Key : , value : Example Test Message 0, Partition 0 Key : , value : Example Test Message 0, Partition 0 Key : , value : Example Test Message 0, Partition 0 Key : , value : Example Test Message 0, Partition 0 Key : , value : Example Test Message 0, Partition 0
Nota
Si ha utilizado la consola para producir un mensaje de prueba, la clave de cada mensaje esNull
Pasos siguientes
Consulte los siguientes recursos para obtener más información: