Alguns de vocês devem ter notado que lançamos o Couchbase 1.8 hoje cedoe um novo conjunto de clientes inteligentes para vários idiomas. Para mim, pessoalmente, isso é um marco, porque libcouchbase agora é um cliente compatível com a linguagem C.
Então, por que me preocupo com isso? Bem, a libcouchbase começou a partir de minha necessidade de testar facilmente vários componentes do servidor. Como fiz a maior parte do meu desenvolvimento nos componentes do servidor implementados em C, fazia sentido usar C para meus testes.
Recebi algumas perguntas sobre como a libcouchbase funciona em um contexto multithread, portanto, devo começar esclarecendo o seguinte: a libcouchbase não usa qualquer O libcouchbase é uma forma de bloqueio para proteger suas estruturas de dados internas, mas isso não significa que você não possa usar o libcouchbase em um programa multithread. Tudo o que isso significa é que você, como usuário cliente, deve usar o bloqueio para se proteger do acesso à instância do libcouchbase de vários threads ao mesmo tempo, ou simplesmente deixar que cada thread opere em sua própria instância do libcouchbase. Uma maneira fácil de resolver isso é ter um "pool" de instâncias do libcouchbase para o qual cada thread pop e push sua instância sempre que precisar acessar um servidor Couchbase. O acesso a esse pool deve ser protegido com um bloqueio (mas acho que você já descobriu isso ;-)
Nesta postagem do blog, criarei um programa de demonstração que você pode usar para fazer upload de documentos JSON em um servidor Couchbase. Você encontrará o código-fonte completo disponível em https://github.com/trondn/vacuum se você quiser experimentar o exemplo.
A ideia desse programa é "monitorar" um diretório e fazer upload de todos os arquivos que aparecem nele para um cluster do Couchbase. Tenho certeza de que a maioria de vocês começou a pensar: "Como podemos fazer isso de forma portátil?". Essa não é uma tarefa fácil, portanto, nem vou tentar fazer isso. Vou tentar escrevê-lo em um semi-portátil de modo que não seja tão difícil implementá-lo em outras plataformas. Isso significa que estou usando as seguintes limitações:
- Estou usando abrir e readdir para percorrer o diretório. Isso pode ser facilmente reimplementado com FindFirst e LocalizarPróximo no Microsoft Windows.
- Monitorar o diretório significa que vou fazer uma varredura no diretório e, em seguida, aguardar um determinado número de segundos antes de executar outra varredura. Sei que algumas plataformas oferecem suporte à assinatura de alterações no sistema de arquivos, mas não vou gastar tempo com isso (pelo menos não agora ;-)).
- Para evitar o bloqueio de arquivos ou o acesso ao arquivo enquanto outros estão gravando o arquivo, os clientes devem gravar o arquivo no diretório com um "ponto" inicial no nome do arquivo e, em seguida, renomear o arquivo quando terminarem. O programa ignora todos os arquivos que começam com um ponto.
Então, vamos pular para o código. O primeiro trecho de código que pode ser interessante de examinar é aquele em que criamos a instância da libcouchbase em main():
Se (instância == NULL) {
fprintf(stderr, "Failed to create couchbase instancen");
exit(EXIT_FAILURE);
}
O trecho de código acima cria a instância da libcouchbase. Não há como usar uma estrutura estática para isso, pois isso dificultará muito a manutenção da compatibilidade binária. Gosto de poder corrigir bugs na biblioteca e lançar novas versões que você possa usar sem precisar recompilar seu programa, e, ao ocultar as estruturas de dados internas dos clientes, fica mais fácil garantir que o cliente não dependa do tamanho delas. O primeiro parâmetro para libcouchbase_create é o nome (e a porta) da porta REST do servidor couchbase (padrão: localhost:8091). O segundo e o terceiro parâmetros são as credenciais que você gostaria de usar para se conectar à porta REST para obter as informações do pool (o padrão é não autenticar). O quarto parâmetro é o bucket ao qual você gostaria de se conectar e, se não especificar um bucket, você acabará no "bucket padrão". O quinto argumento é um objeto especial que você pode querer usar se for utilizar recursos "avançados" da libcouchbase. A maioria dos usuários provavelmente usará apenas os padrões e passará NULL aqui.
A próxima coisa que precisamos fazer é configurar alguns manipuladores de retorno de chamada para podermos descobrir o que acontece. No exemplo, usaremos apenas uma operação (para carregar dados no cache), portanto, precisaremos configurar um manipulador para capturar o resultado das operações de armazenamento. Infelizmente, também podemos encontrar problemas, portanto, precisamos configurar um manipulador de erros (voltaremos ao trabalho daqui a pouco).
libcouchbase_set_error_callback(instance, error_callback);
Agora que criamos e inicializamos a instância, precisamos tentar nos conectar ao cluster do Couchbase:
se (ret != LIBCOUCHBASE_SUCCESS) {
fprintf(stderr, "Falha na conexão: %sn",
libcouchbase_strerror(instance, ret));
exit(EXIT_FAILURE);
}
Devido ao fato de a libcouchbase ser totalmente assíncrona, tudo o que aconteceu acima foi que iniciamos a conexão. Isso significa que precisamos esperar para que o servidor seja conectado ao cluster do Couchbase e se conecte ao bucket correto. Se o nosso programa tivesse que fazer outras coisas, agora seria o momento de fazê-lo, mas como não temos nenhuma outra inicialização a fazer, podemos simplesmente esperar que ele seja concluído:
Um dos recursos "legais" que temos na libcouchbase é que ela fornece uma interface de estatísticas internas, portanto, podemos dizer a ela para coletar informações de tempo das operações com o seguinte trecho:
Se ((ret = libcouchbase_enable_timings(instance) != LIBCOUCHBASE_SUCCESS)) {
fprintf(stderr, "Failed to enable timings: %sn",
libcouchbase_strerror(instance, ret));
}
Nosso programa agora está totalmente inicializado e podemos entrar no loop principal, que se parece bastante com isso:
{
process_files();
sleep(nsec);
}
Então, como o nosso process_files() como é? Não vou deixar o exemplo muito grande colando-o todo, mas a primeira parte parece:
Se (strcmp(de->d_name, ".dump_stats") == 0) {
fprintf(stdout, "Dumping stats:n");
libcouchbase_get_timings(instance, stdout, timings_callback);
fprintf(stdout, "--n");
remove(de->d_name);<
}
continuar;
}
Como você pode ver no trecho de código acima, ignoraremos todos os arquivos que começam com '.', exceto pelo arquivo chamado ".dump_stats". Sempre que vemos esse arquivo, despejamos os tempos das estatísticas internas usando o comando timings_callback (Voltarei a esse assunto mais tarde).
A próxima coisa que faremos é tentar ler o arquivo na memória e decodificar seu JSON antes de tentarmos obter o "_id" para usar como chave. Se tudo isso for bem-sucedido, tentaremos armazenar os dados no Coucbase com:
ret = libcouchbase_store(instance, &error, LIBCOUCHBASE_SET,
id->valuestring, strlen(id->valuestring),
ptr, size, 0, 0, 0);
se (ret == LIBCOUCHBASE_SUCCESS) {
libcouchbase_wait(instance);
} else {
erro = 1;
}
A parte do &error aqui é bastante interessante. É um "cookie" passado para o retorno de chamada, para que eu possa saber se encontrei um problema ou não. Você verá como estou usando isso quando discutir a função retorno de armazenamento abaixo.
Essa é basicamente toda a lógica importante do exemplo. Prometi que voltaria a falar sobre os diferentes retornos de chamada, portanto, vamos começar examinando o retorno de chamada de erro:
Erro da libcouchbase_error_t,
const char *errinfo)
{
/* Ignorar tempos limite... */
se (erro != LIBCOUCHBASE_ETIMEDOUT) {
fprintf(stderr, "rFATAL ERROR: %sn",
libcouchbase_strerror(instance, error));
Se (errinfo && strlen(errinfo) != 0) {
fprintf(stderr, "t "%s "n", errinfo);
}
exit(EXIT_FAILURE);
}
}
Como você pode ver no snippet acima, a libcouchbase chamará a função retorno_de_erro sempre que ocorrer um tempo limite, mas queremos apenas tentar novamente a operação. Se encontrarmos um erro real, imprimiremos uma mensagem de erro e encerraremos o programa.
O próximo retorno de chamada que usamos é o retorno de armazenamento. Ela é chamada quando a operação de armazenamento é concluída, portanto, é o lugar certo para descobrirmos se ocorreu algum erro durante o armazenamento dos dados. Nosso retorno de chamada tem a seguinte aparência:
const void *cookie,
libcouchbase_storage_t operation,
libcouchbase_error_t err,
const void *key, size_t nkey,
uint64_t cas)
{
int *error = (void*)cookie;
se (err == LIBCOUCHBASE_SUCCESS) {
*erro = 0;
} else {
*Erro = 1;
fprintf(stderr, "Falha ao armazenar "");
fwrite(key, 1, nkey, stderr);
fprintf(stderr, "": %sn",
libcouchbase_strerror(instance, err));
fflush(stderr);
}
}
Como você pode ver, estamos armazenando o resultado da operação no número inteiro passado como cookie. O leitor atento pode perceber que também poderíamos desvincular o arquivo e remover a memória da chamada de retorno (se, em vez disso, tivéssemos fornecido essas informações como cookie ;))
A última chamada de retorno a ser abordada é a chamada de retorno de tempo que estamos usando para descarregar as estatísticas de tempo.
static void timings_callback(libcouchbase_t instance, const void *cookie,
libcouchbase_timeunit_t timeunit,
uint32_t min, uint32_t max,
uint32_t total, uint32_t maxtotal)
{
char buffer[1024];
int offset = sprintf(buffer, "[%3u - %3u]", min, max);
switch (timeunit) {
case LIBCOUCHBASE_TIMEUNIT_NSEC:
offset += sprintf(buffer + offset, "ns");
pausa;
case LIBCOUCHBASE_TIMEUNIT_USEC:
offset += sprintf(buffer + offset, "us");
pausa;
case LIBCOUCHBASE_TIMEUNIT_MSEC:
offset += sprintf(buffer + offset, "ms");
pausa;
case LIBCOUCHBASE_TIMEUNIT_SEC:
offset += sprintf(buffer + offset, "s");
pausa;
padrão:
;
}
int num = (float)40.0 * (float)total / (float)maxtotal;
offset += sprintf(buffer + offset, " |");
for (int ii = 0; ii < num; ++ii) {
offset += sprintf(buffer + offset, "#");
}
offset += sprintf(buffer + offset, " - %un", total);
fputs(buffer, (FILE*)cookie);
}
Quando você solicita os tempos da libcouchbase, ela informa todas as métricas de tempo coletadas chamando o retorno de chamada de tempos. Como você pode ver na API, você obterá o valor mínimo e máximo do intervalo e o número de operações realizadas dentro desse intervalo. Essas métricas não devem ser consideradas como números exatos, pois dependem do que você faz no código do cliente desde o momento em que chama a operação até chamar libcouchbase_wait para que a operação seja concluída.
Então, vamos executar o programa. Eu preenchi previamente /var/spool/vacuum com vários arquivos JSON, para que o programa faça alguma coisa.
dormindo 3 segundos antes de tentar novamente.
Em outro withdow, executo o comando:
E quando o cronômetro expira na primeira janela, ele é impresso:
[ 60 - 69]us |######################################## - 18
[ 70 - 79]us |## - 1
[240 - 249]us |## - 1
—-
dormindo 3 segundos antes de tentar novamente.
Espero que este blog tenha revelado como é fácil usar a libcouchbase para se comunicar com um cluster do Couchbase. Temos vários clientes para outras linguagens de programação, como PHP e Ruby, criados com base na libcouchbase, portanto, posso prometer que você verá mais funcionalidades adicionadas!
