Kafka通讯代码02

  • Producer
  • Consumer
  • GroupConsumer

1、MqConsumer.java

package com.neohope.kafka.test;

import kafka.api.FetchRequest;
import kafka.api.FetchRequestBuilder;
import kafka.api.PartitionOffsetRequestInfo;
import kafka.common.ErrorMapping;
import kafka.common.TopicAndPartition;
import kafka.javaapi.*;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.message.MessageAndOffset;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class MqConsumer {
    private List<String> m_replicaBrokers = new ArrayList<String>();

    public MqConsumer() {
        m_replicaBrokers = new ArrayList<String>();
    }

    public void run(String a_topic, int a_partition, List<String> a_seedBrokers, int a_port) throws Exception {
        // find the meta data about the topic and partition we are interested in
        //
        PartitionMetadata metadata = findLeader(a_seedBrokers, a_port, a_topic, a_partition);
        if (metadata == null) {
            System.out.println("Can't find metadata for Topic and Partition. Exiting");
            return;
        }
        if (metadata.leader() == null) {
            System.out.println("Can't find Leader for Topic and Partition. Exiting");
            return;
        }
        String leadBroker = metadata.leader().host();
        String clientName = "Client_" + a_topic + "_" + a_partition;

        SimpleConsumer consumer = new SimpleConsumer(leadBroker, a_port, 100000, 64 * 1024, clientName);
        long readOffset = getLastOffset(consumer,a_topic, a_partition, kafka.api.OffsetRequest.EarliestTime(), clientName);

        boolean bEnd =false;
        while (!bEnd) {
            if (consumer == null) {
                consumer = new SimpleConsumer(leadBroker, a_port, 100000, 64 * 1024, clientName);
            }
            FetchRequest req = new FetchRequestBuilder()
                    .clientId(clientName)
                    .addFetch(a_topic, a_partition, readOffset, 100000) // Note: this fetchSize of 100000 might need to be increased if large batches are written to Kafka
                    .build();
            FetchResponse fetchResponse = consumer.fetch(req);

            if (fetchResponse.hasError()) {
                // Something went wrong!
                short code = fetchResponse.errorCode(a_topic, a_partition);
                System.out.println("Error fetching data from the Broker:" + leadBroker + " Reason: " + code);
                if (code == ErrorMapping.OffsetOutOfRangeCode())  {
                    // We asked for an invalid offset. For simple case ask for the last element to reset
                    readOffset = getLastOffset(consumer,a_topic, a_partition, kafka.api.OffsetRequest.LatestTime(), clientName);
                    continue;
                }
                consumer.close();
                consumer = null;
                leadBroker = findNewLeader(leadBroker, a_topic, a_partition, a_port);
                continue;
            }

            for (MessageAndOffset messageAndOffset : fetchResponse.messageSet(a_topic, a_partition)) {
                long currentOffset = messageAndOffset.offset();

                if (currentOffset < readOffset) {
                    System.out.println("Found an old offset: " + currentOffset + " Expecting: " + readOffset);
                    continue;
                }
                readOffset = messageAndOffset.nextOffset();
                ByteBuffer key = messageAndOffset.message().key();
                byte[] bytesKey = new byte[key.limit()];
                key.get(bytesKey);
                String szKey = new String(bytesKey, "UTF-8");
                if(szKey.equals("-=END=-"))bEnd =true;

                ByteBuffer payload = messageAndOffset.message().payload();
                byte[] bytesPayload = new byte[payload.limit()];
                payload.get(bytesPayload);
                String szPaylaod = new String(bytesPayload, "UTF-8");

                String offset = String.valueOf(messageAndOffset.offset());

                System.out.println("offset=" + offset+ " key=" + szKey +" value="+szPaylaod);
            }

            Thread.sleep(1000);
        }
        if (consumer != null) consumer.close();
    }

    public static long getLastOffset(SimpleConsumer consumer, String topic, int partition,
                                     long whichTime, String clientName) {
        TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partition);
        Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo = new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
        requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(whichTime, 1));
        kafka.javaapi.OffsetRequest request = new kafka.javaapi.OffsetRequest(
                requestInfo, kafka.api.OffsetRequest.CurrentVersion(), clientName);
        OffsetResponse response = consumer.getOffsetsBefore(request);

        if (response.hasError()) {
            System.out.println("Error fetching data Offset Data the Broker. Reason: " + response.errorCode(topic, partition) );
            return 0;
        }
        long[] offsets = response.offsets(topic, partition);
        return offsets[0];
    }

    private String findNewLeader(String a_oldLeader, String a_topic, int a_partition, int a_port) throws Exception {
        for (int i = 0; i < 3; i++) {
            boolean goToSleep = false;
            PartitionMetadata metadata = findLeader(m_replicaBrokers, a_port, a_topic, a_partition);
            if (metadata == null) {
                goToSleep = true;
            } else if (metadata.leader() == null) {
                goToSleep = true;
            } else if (a_oldLeader.equalsIgnoreCase(metadata.leader().host()) && i == 0) {
                // first time through if the leader hasn't changed give ZooKeeper a second to recover
                // second time, assume the broker did recover before failover, or it was a non-Broker issue
                //
                goToSleep = true;
            } else {
                return metadata.leader().host();
            }
            if (goToSleep) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ie) {
                }
            }
        }
        System.out.println("Unable to find new leader after Broker failure. Exiting");
        throw new Exception("Unable to find new leader after Broker failure. Exiting");
    }

    private PartitionMetadata findLeader(List<String> a_seedBrokers, int a_port, String a_topic, int a_partition) {
        PartitionMetadata returnMetaData = null;
        loop:
        for (String seed : a_seedBrokers) {
            SimpleConsumer consumer = null;
            try {
                consumer = new SimpleConsumer(seed, a_port, 100000, 64 * 1024, "leaderLookup");
                List<String> topics = Collections.singletonList(a_topic);
                TopicMetadataRequest req = new TopicMetadataRequest(topics);
                kafka.javaapi.TopicMetadataResponse resp = consumer.send(req);

                List<TopicMetadata> metaData = resp.topicsMetadata();
                for (TopicMetadata item : metaData) {
                    for (PartitionMetadata part : item.partitionsMetadata()) {
                        if (part.partitionId() == a_partition) {
                            returnMetaData = part;
                            break loop;
                        }
                    }
                }
            } catch (Exception e) {
                System.out.println("Error communicating with Broker [" + seed + "] to find Leader for [" + a_topic
                        + ", " + a_partition + "] Reason: " + e);
            } finally {
                if (consumer != null) consumer.close();
            }
        }
        if (returnMetaData != null) {
            m_replicaBrokers.clear();
            for (kafka.cluster.BrokerEndPoint replica : returnMetaData.replicas()) {
                m_replicaBrokers.add(replica.host());
            }
        }
        return returnMetaData;
    }

    public static void main(String args[]) {
        MqConsumer mqc = new MqConsumer();

        List<String> seeds = new ArrayList<String>();
        seeds.add("localhost");

        String topic = "neoTopic";
        int partition = 0;
        int port = 9092;

        try {
            mqc.run(topic, partition, seeds, port);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

Continue reading Kafka通讯代码02

Kafka通讯代码01

  • Producer
  • Consumer
  • GroupConsumer

1、MqProducer.java

package com.neohope.kafka.test;

import java.util.*;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class MqProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("metadata.broker.list", "localhost:9092");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("partitioner.class", "com.neohope.kafka.test.SimplePartitioner");
        props.put("request.required.acks", "1");

        ProducerConfig config = new ProducerConfig(props);
        Producer<String, String> producer = new Producer<String, String>(config);
        for(int i=0;i<100;i++) {
            KeyedMessage<String, String> data = new KeyedMessage<String, String>("neoTopic", "key"+i, "value"+i);
            producer.send(data);
        }
        KeyedMessage<String, String> data = new KeyedMessage<String, String>("neoTopic", "-=END=-", "-=END=-");
        producer.send(data);

        producer.close();
    }
}

Continue reading Kafka通讯代码01

ActiveMQ通讯代码04

  • JMS方式调用
  • Queue方式调用
  • Topic方式调用
  • ReqRsp方式调用

1、TestMsg.java

package com.neohope.ActiveMQ.test.beans;

public class TestMsg implements java.io.Serializable{
    private static final long serialVersionUID = 12345678;

    public TestMsg(int taskId, String taskInfo, int taskLevel) {
        this.taskId = taskId;
        this.taskInfo = taskInfo;
        this.taskLevel = taskLevel;
    }

    public int taskId;
    public String taskInfo;
    public int taskLevel;
}

Continue reading ActiveMQ通讯代码04

ActiveMQ通讯代码03

  • JMS方式调用
  • Queue方式调用
  • Topic方式调用
  • ReqRsp方式调用

1、TestMsg.java

package com.neohope.ActiveMQ.test.beans;

public class TestMsg implements java.io.Serializable{
    private static final long serialVersionUID = 12345678;

    public TestMsg(int taskId, String taskInfo, int taskLevel) {
        this.taskId = taskId;
        this.taskInfo = taskInfo;
        this.taskLevel = taskLevel;
    }

    public int taskId;
    public String taskInfo;
    public int taskLevel;
}

Continue reading ActiveMQ通讯代码03

ActiveMQ通讯代码02

  • JMS方式调用
  • Queue方式调用
  • Topic方式调用
  • ReqRsp方式调用

1、TestMsg.java

package com.neohope.ActiveMQ.test.beans;

public class TestMsg implements java.io.Serializable{
    private static final long serialVersionUID = 12345678;

    public TestMsg(int taskId, String taskInfo, int taskLevel) {
        this.taskId = taskId;
        this.taskInfo = taskInfo;
        this.taskLevel = taskLevel;
    }

    public int taskId;
    public String taskInfo;
    public int taskLevel;
}

Continue reading ActiveMQ通讯代码02

ActiveMQ通讯代码01

  • JMS方式调用
  • Queue方式调用
  • Topic方式调用
  • ReqRsp方式调用

1、TestMsg.java

package com.neohope.ActiveMQ.test.beans;

public class TestMsg implements java.io.Serializable{
    private static final long serialVersionUID = 12345678;

    public TestMsg(int taskId, String taskInfo, int taskLevel) {
        this.taskId = taskId;
        this.taskInfo = taskInfo;
        this.taskLevel = taskLevel;
    }

    public int taskId;
    public String taskInfo;
    public int taskLevel;
}

Continue reading ActiveMQ通讯代码01

Memcached常用操作

*生产环境建议直接用linux

1、命令行直接运行
1.1、可以直接指定参数运行

#最大16M内存,监听11211端口,最大连接数8
memcached.exe -m 16 -p 11211 -c 8

1.2、可以注册为Windows服务,再运行

#注册为服务
memcached.exe -d install
#开启服务
memcached.exe -d start
#关闭服务
memcached.exe -d stop
#卸载服务
memcached.exe -d uninstall

Continue reading Memcached常用操作

DBUS发送接收数据(下)

1、编译后,由于没有进行配置,默认是无法运行的。
为了可以正常运行,增加或修改下面的配置文件即可。
/etc/dbus-1.0/system-local.conf

<!-- This configuration file controls the systemwide message bus.
     Add a system-local.conf and edit that rather than changing this 
     file directly. -->

<!-- Note that there are any number of ways you can hose yourself
     security-wise by screwing up this file; in particular, you
     probably don't want to listen on any more addresses, add any more
     auth mechanisms, run as a different user, etc. -->

<!DOCTYPE busconfig PUBLIC "-//freedesktop//DTD D-Bus Bus Configuration 1.0//EN"
 "http://www.freedesktop.org/standards/dbus/1.0/busconfig.dtd">
<busconfig>

  <!-- Our well-known bus type, do not change this -->
  <type>system</type>

  <!-- Run as special user -->
  <user>messagebus</user>

  <!-- Fork into daemon mode -->
  <fork/>

  <!-- We use system service launching using a helper -->
  <standard_system_servicedirs/>

  <!-- This is a setuid helper that is used to launch system services -->
  <servicehelper>/usr/lib/dbus-1.0/dbus-daemon-launch-helper</servicehelper>

  <!-- Write a pid file -->
  <pidfile>/var/run/dbus/pid</pidfile>

  <!-- Enable logging to syslog -->
  <syslog/>

  <!-- Only allow socket-credentials-based authentication -->
  <auth>EXTERNAL</auth>

  <!-- Only listen on a local socket. (abstract=/path/to/socket 
       means use abstract namespace, don't really create filesystem 
       file; only Linux supports this. Use path=/whatever on other 
       systems.) -->
  <listen>unix:path=/var/run/dbus/system_bus_socket</listen>

  <policy context="default">
    <!-- All users can connect to system bus -->
    <allow user="*"/>

    <!-- Holes must be punched in service configuration files for
         name ownership and sending method calls -->
    <allow own="*"/>
    <allow send_type="method_call"/>

    <!-- Signals and reply messages (method returns, errors) are allowed
         by default -->
    <allow send_type="signal"/>
    <allow send_requested_reply="true" send_type="method_return"/>
    <allow send_requested_reply="true" send_type="error"/>

    <!-- All messages may be received by default -->
    <allow receive_type="method_call"/>
    <allow receive_type="method_return"/>
    <allow receive_type="error"/>
    <allow receive_type="signal"/>

    <!-- Allow anyone to talk to the message bus -->
    <allow send_destination="org.freedesktop.DBus"/>
    <!-- But disallow some specific bus services -->
    <deny send_destination="org.freedesktop.DBus"
          send_interface="org.freedesktop.DBus"
          send_member="UpdateActivationEnvironment"/>
    <deny send_destination="org.freedesktop.DBus"
          send_interface="org.freedesktop.systemd1.Activator"/>
  </policy>

  <!-- Only systemd, which runs as root, may report activation failures. -->
  <policy user="root">
    <allow send_destination="org.freedesktop.DBus"
           send_interface="org.freedesktop.systemd1.Activator"/>
  </policy>

  <!-- Config files are placed here that among other things, punch 
       holes in the above policy for specific services. -->
  <includedir>system.d</includedir>

  <include if_selinux_enabled="yes" selinux_root_relative="yes">contexts/dbus_contexts</include>

</busconfig>

2、测试消息发送
服务端:

$./testdbus_s.bin receive
Listening for signals
Match rule sent
Got Signal with value: Hello
Got Signal with value: Hi
Bye......

客户端:

$./testdbus_s.bin send Hello
Sending signal with value: Hello
Signal Sent

$./testdbus_s.bin send Hi
Sending signal with value: Hi
Signal Sent

$./testdbus_s.bin send Bye
Sending signal with value: Bye
Signal Sent

3、测试方法调用
服务端:

$./testdbus_s.bin listen
Listening for method calls
Method Invoked with value: Hello
Method Invoked with value: Hi
Bye......

客户端:

$./testdbus_s.bin query Hello
Calling remote method with Hello
Request Sent
Got Reply: 1, 21614

$./testdbus_s.bin query Hi
Calling remote method with Hi
Request Sent
Got Reply: 1, 21614

$./testdbus_s.bin query Bye
Calling remote method with Bye
Request Sent
Got Reply: 1, 21614

DBUS发送接收数据(上)

1、Server端
testdbus_s.c

#define DBUS_API_SUBJECT_TO_CHANGE
#include <dbus/dbus.h>
#include <stdbool.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>

/**
 * Wait for signals on the bus and reply
 */
void receive()
{
  DBusMessage* msg;
  DBusMessageIter args;
  DBusConnection* conn;
  DBusError err;
  int ret;
  char* sigvalue;
  int loop=1;

  printf("Listening for signals\n");

  // initialise the errors
  dbus_error_init(&err);
  
  // connect to the bus and check for errors
  conn = dbus_bus_get(DBUS_BUS_SYSTEM, &err);
  if (dbus_error_is_set(&err)) { 
     fprintf(stderr, "Connection Error (%s)\n", err.message);
     dbus_error_free(&err); 
  }
  if (NULL == conn) { 
     exit(1);
  }
  
  // request our name on the bus and check for errors
  ret = dbus_bus_request_name(conn, "neohope.dbus.signal.target", DBUS_NAME_FLAG_REPLACE_EXISTING , &err);
  if (dbus_error_is_set(&err)) { 
     fprintf(stderr, "Name Error (%s)\n", err.message);
     dbus_error_free(&err); 
  }
  if (DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER != ret) {
     exit(1);
  }

  // add a rule for which messages we want to see
  dbus_bus_add_match(conn, "type='signal',interface='neohope.dbus.signal.Type'", &err); // see signals from the given interface
  dbus_connection_flush(conn);
  if (dbus_error_is_set(&err)) { 
     fprintf(stderr, "Match Error (%s)\n", err.message);
     exit(1); 
  }
  printf("Match rule sent\n");

  // loop listening for signals being emmitted
  while (loop) {

     // non blocking read of the next available message
     dbus_connection_read_write(conn, 0);
     msg = dbus_connection_pop_message(conn);

     // loop again if we haven't read a message
     if (NULL == msg) { 
        sleep(1);
        continue;
     }

     // check if the message is a signal from the correct interface and with the correct name
     if (dbus_message_is_signal(msg, "neohope.dbus.signal.Type", "TestSignal")) {
        
        // read the parameters
        if (!dbus_message_iter_init(msg, &args))
           fprintf(stderr, "Message Has No Parameters\n");
        else if (DBUS_TYPE_STRING != dbus_message_iter_get_arg_type(&args)) 
           fprintf(stderr, "Argument is not string!\n"); 
        else
           dbus_message_iter_get_basic(&args, &sigvalue);
        
        if(strcmp("Bye",sigvalue)==0) {
           loop = 0;
           printf("Bye......\n");
        }
        else {
          printf("Got Signal with value: %s\n", sigvalue);
        }
     }

     // free the message
     dbus_message_unref(msg);
  }
  // do not close the connection
  // dbus_connection_close(conn);
}

/**
 * Deal with remote method call 
 */
int reply_to_method_call(DBusMessage* msg, DBusConnection* conn)
{
  DBusMessage* reply;
  DBusMessageIter args;
  int stat = 1;
  int ret = 1;
  dbus_uint32_t level = 21614;
  dbus_uint32_t serial = 0;
  char* param = "";

  // read the arguments
  if (!dbus_message_iter_init(msg, &args))
     fprintf(stderr, "Message has no arguments!\n"); 
  else if (DBUS_TYPE_STRING != dbus_message_iter_get_arg_type(&args)) 
     fprintf(stderr, "Argument is not string!\n"); 
  else 
     dbus_message_iter_get_basic(&args, &param);

  if(strcmp("Bye",param)==0){
    ret = 0;
    printf ("Bye......\n");
  }
  else {
    printf("Method Invoked with value: %s\n", sigvalue);
  }

  // create a reply from the message
  reply = dbus_message_new_method_return(msg);

  // add the arguments to the reply
  dbus_message_iter_init_append(reply, &args);
  if (!dbus_message_iter_append_basic(&args, DBUS_TYPE_BOOLEAN, &stat)) { 
     fprintf(stderr, "Out Of Memory!\n"); 
     exit(1);
  }
  if (!dbus_message_iter_append_basic(&args, DBUS_TYPE_UINT32, &level)) { 
     fprintf(stderr, "Out Of Memory!\n"); 
     exit(1);
  }

  // send the reply && flush the connection
  if (!dbus_connection_send(conn, reply, &serial)) {
     fprintf(stderr, "Out Of Memory!\n"); 
     exit(1);
  }
  dbus_connection_flush(conn);

  // free the reply
  dbus_message_unref(reply);
  
  return ret;
}

/**
 * Server that exposes a method call and waits for it to be called
 */
void listen() 
{
  DBusMessage* msg;
  DBusMessage* reply;
  DBusMessageIter args;
  DBusConnection* conn;
  DBusError err;
  int loop = 1;
  int ret;
  char* param;

  printf("Listening for method calls\n");

  // initialise the error
  dbus_error_init(&err);
  
  // connect to the bus and check for errors
  conn = dbus_bus_get(DBUS_BUS_SYSTEM, &err);
  if (dbus_error_is_set(&err)) { 
     fprintf(stderr, "Connection Error (%s)\n", err.message); 
     dbus_error_free(&err); 
  }
  if (NULL == conn) {
     fprintf(stderr, "Connection Null\n"); 
     exit(1); 
  }
  
  // request our name on the bus and check for errors
  ret = dbus_bus_request_name(conn, "neohope.dbus.method.provider", DBUS_NAME_FLAG_REPLACE_EXISTING , &err);
  if (dbus_error_is_set(&err)) { 
     fprintf(stderr, "Name Error (%s)\n", err.message); 
     dbus_error_free(&err);
  }
  if (DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER != ret) { 
     fprintf(stderr, "Not Primary Owner (%d)\n", ret);
     exit(1); 
  }

  // loop, testing for new messages
  while (loop) {
     // non blocking read of the next available message
     dbus_connection_read_write(conn, 0);
     msg = dbus_connection_pop_message(conn);

     // loop again if we haven't got a message
     if (NULL == msg) { 
        sleep(1); 
        continue; 
     }
     
     // check this is a method call for the right interface & method
     if (dbus_message_is_method_call(msg, "neohope.dbus.method.Type", "TestMethod")) 
        loop = reply_to_method_call(msg, conn);

     // free the message
     dbus_message_unref(msg);
  }

  // do not close the connection
  // dbus_connection_close(conn);
}

int main(int argc, char** argv)
{
  if (2 > argc) {
     printf ("Syntax: testdbus_s [receive|listen] [<param>]\n");
     return 1;
  }
  char* param = "no param";
  if (3 >= argc && NULL != argv[2]) param = argv[2];

  if (0 == strcmp(argv[1], "receive"))
     receive();
  else if (0 == strcmp(argv[1], "listen"))
     listen();
  else {
     printf ("Syntax: testdbus_s [receive|listen] [<param>]\n");
     return 1;
  }
  return 0;
}

2、Client端
testdbus_c.c

#define DBUS_API_SUBJECT_TO_CHANGE
#include <dbus/dbus.h>
#include <stdbool.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>

/**
 * Connect to the DBUS bus and send a broadcast signal
 */
void sendsignal(char* sigvalue)
{
  DBusMessage* msg;
  DBusMessageIter args;
  DBusConnection* conn;
  DBusError err;
  int ret;
  dbus_uint32_t serial = 0;

  printf("Sending signal with value: %s\n", sigvalue);

  // initialise the error value
  dbus_error_init(&err);

  // connect to the DBUS system bus, and check for errors
  conn = dbus_bus_get(DBUS_BUS_SYSTEM, &err);
  if (dbus_error_is_set(&err)) { 
     fprintf(stderr, "Connection Error (%s)\n", err.message); 
     dbus_error_free(&err); 
  }
  if (NULL == conn) { 
     exit(1); 
  }

  // register our name on the bus, and check for errors
  ret = dbus_bus_request_name(conn, "neohope.dbus.signal.source", DBUS_NAME_FLAG_REPLACE_EXISTING , &err);
  if (dbus_error_is_set(&err)) { 
     fprintf(stderr, "Name Error (%s)\n", err.message); 
     dbus_error_free(&err); 
  }
  if (DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER != ret) { 
     exit(1);
  }

  // create a signal & check for errors 
  msg = dbus_message_new_signal("/neohope/dbus/signal/Object", // object name of the signal
                                "neohope.dbus.signal.Type", // interface name of the signal
                                "TestSignal"); // name of the signal
  if (NULL == msg) 
  { 
     fprintf(stderr, "Message Null\n"); 
     exit(1); 
  }

  // append arguments onto signal
  dbus_message_iter_init_append(msg, &args);
  if (!dbus_message_iter_append_basic(&args, DBUS_TYPE_STRING, &sigvalue)) {
     fprintf(stderr, "Out Of Memory!\n"); 
     exit(1);
  }

  // send the message and flush the connection
  if (!dbus_connection_send(conn, msg, &serial)) {
     fprintf(stderr, "Out Of Memory!\n"); 
     exit(1);
  }
  dbus_connection_flush(conn);
  
  printf("Signal Sent\n");
  
  // free the message
  dbus_message_unref(msg);
  // do not close the connection
  // dbus_connection_close(conn);
}

/**
 * Call a method on a remote object
 */
void query(char* param) 
{
  DBusMessage* msg;
  DBusMessageIter args;
  DBusConnection* conn;
  DBusError err;
  DBusPendingCall* pending;
  int ret;
  int stat;
  dbus_uint32_t level;

  printf("Calling remote method with %s\n", param);

  // initialiset the errors
  dbus_error_init(&err);

  // connect to the system bus and check for errors
  conn = dbus_bus_get(DBUS_BUS_SYSTEM, &err);
  if (dbus_error_is_set(&err)) { 
     fprintf(stderr, "Connection Error (%s)\n", err.message); 
     dbus_error_free(&err);
  }
  if (NULL == conn) { 
     exit(1); 
  }

  // request our name on the bus
  ret = dbus_bus_request_name(conn, "neohope.dbus.method.caller", DBUS_NAME_FLAG_REPLACE_EXISTING , &err);
  if (dbus_error_is_set(&err)) { 
     fprintf(stderr, "Name Error (%s)\n", err.message); 
     dbus_error_free(&err);
  }
  if (DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER != ret) { 
     exit(1);
  }

  // create a new method call and check for errors
  msg = dbus_message_new_method_call("neohope.dbus.method.provider", // target for the method call
                                     "/neohope/dbus/method/Object", // object to call on
                                     "neohope.dbus.method.Type", // interface to call on
                                     "TestMethod"); // method name
  if (NULL == msg) { 
     fprintf(stderr, "Message Null\n");
     exit(1);
  }

  // append arguments
  dbus_message_iter_init_append(msg, &args);
  if (!dbus_message_iter_append_basic(&args, DBUS_TYPE_STRING, &param)) {
     fprintf(stderr, "Out Of Memory!\n"); 
     exit(1);
  }
  
  // send message and get a handle for a reply
  if (!dbus_connection_send_with_reply (conn, msg, &pending, -1)) { // -1 is default timeout
     fprintf(stderr, "Out Of Memory!\n"); 
     exit(1);
  }
  if (NULL == pending) { 
     fprintf(stderr, "Pending Call Null\n"); 
     exit(1); 
  }
  dbus_connection_flush(conn);
  
  printf("Request Sent\n");
  
  // free message
  dbus_message_unref(msg);
  
  // block until we recieve a reply
  dbus_pending_call_block(pending);

  // get the reply message
  msg = dbus_pending_call_steal_reply(pending);
  if (NULL == msg) {
     fprintf(stderr, "Reply Null\n"); 
     exit(1); 
  }
  // free the pending message handle
  dbus_pending_call_unref(pending);

  // read the parameters
  if (!dbus_message_iter_init(msg, &args))
     fprintf(stderr, "Message has no arguments!\n"); 
  else if (DBUS_TYPE_BOOLEAN != dbus_message_iter_get_arg_type(&args)) 
     fprintf(stderr, "Argument is not boolean!\n"); 
  else
     dbus_message_iter_get_basic(&args, &stat);

  if (!dbus_message_iter_next(&args))
     fprintf(stderr, "Message has too few arguments!\n"); 
  else if (DBUS_TYPE_UINT32 != dbus_message_iter_get_arg_type(&args)) 
     fprintf(stderr, "Argument is not int!\n"); 
  else
     dbus_message_iter_get_basic(&args, &level);

  printf("Got Reply: %d, %d\n", stat, level);
  
  // free reply and close connection
  dbus_message_unref(msg);   
  //dbus_connection_close(conn);
}

int main(int argc, char** argv)
{
  if (2 > argc) {
     printf ("Syntax: testdbus_c [send|query] [<param>]\n");
     return 1;
  }
  char* param = "no param";
  if (3 >= argc && NULL != argv[2]) param = argv[2];

  if (0 == strcmp(argv[1], "send"))
     sendsignal(param);
  else if (0 == strcmp(argv[1], "query"))
     query(param);
  else {
     printf ("Syntax: testdbus_c [send|query] [<param>]\n");
     return 1;
  }
  return 0;
}

3、Makefile

CC=gcc
LDFLAGS+=-ldbus-1
CFLAGS+=-I/usr/include/dbus-1.0 -I/usr/lib/x86_64-linux-gnu/dbus-1.0/include/

all:testdbus_c.bin testdbus_s.bin

testdbus_c.bin:testdbus_c.o
	$(CC) $(LDFLAGS) -o testdbus_c.bin testdbus_c.o 

testdbus_c.o:testdbus_c.c
	$(CC) $(CFLAGS) -c -o testdbus_c.o testdbus_c.c 

testdbus_s.bin:testdbus_s.o
	$(CC) $(LDFLAGS) -o testdbus_s.bin testdbus_s.o 

testdbus_s.o:testdbus_s.c
	$(CC) $(CFLAGS) -c -o testdbus_s.o testdbus_s.c 

clean:
	rm *.o *.bin

4、编译

make